Checkpoint. Use event timers rather than time.sleep(), etc.

svn path=/rpkid/rpki/left_right.py; revision=2445

4 files changed, 47 insertions, 39 deletions

diff --git a/rpkid/rpki/left_right.py b/rpkid/rpki/left_right.py
index 64852a06..b73102e7 100644
--- a/rpkid/rpki/left_right.py
+++ b/rpkid/rpki/left_right.py
@@ -184,10 +184,12 @@ class self_elt(data_elt):
         def class_loop(class_iterator, rc):
 
           def class_update_failed(e):
+            rpki.log.error(traceback.format_exc())
             rpki.log.warn("Couldn't update class, skipping: %s" % e)
             class_iterator()
 
           def class_create_failed(e):
+            rpki.log.error(traceback.format_exc())
             rpki.log.warn("Couldn't create class, skipping: %s" % e)
             class_iterator()
 
@@ -212,6 +214,7 @@ class self_elt(data_elt):
         rpki.async.iterator(r_msg.payload.classes, class_loop, class_done)
 
       def list_failed(e):
+        rpki.log.error(traceback.format_exc())
         rpki.log.warn("Couldn't get resource class list from parent %r, skipping: %s" % (parent, e))
 
       rpki.up_down.list_pdu.query(parent, got_list, list_failed)
@@ -247,6 +250,7 @@ class self_elt(data_elt):
               rpki.log.debug("Need to reissue child certificate SKI %s" % child_cert.cert.gSKI())
 
               def reissue_failed(e):
+                rpki.log.error(traceback.format_exc())
                 rpki.log.warn("Couldn't reissue child_cert %r, skipping: %s" % (child_cert, e))
                 iterator2()
 
@@ -269,10 +273,12 @@ class self_elt(data_elt):
                 repository.withdraw(child_cert.cert, child_cert.uri(ca), iterator2, withdraw_failed)
 
               def manifest_failed(e):
+                rpki.log.error(traceback.format_exc())
                 rpki.log.warn("Couldn't reissue manifest for %r, skipping: %s" % (ca_detail, e))
                 iterator2()
 
               def withdraw_failed(e):
+                rpki.log.error(traceback.format_exc())
                 rpki.log.warn("Couldn't withdraw old child_cert %r, skipping: %s" % (child_cert, e))
                 iterator2()
 
@@ -284,6 +290,7 @@ class self_elt(data_elt):
         rpki.async.iterator(child_certs, loop2, iterator1)
 
       def irdb_lookup_failed(e):
+        rpki.log.error(traceback.format_exc())
         rpki.log.warn("Couldn't look up child's resources in IRDB, skipping child %r: %s" % (child, e))
         iterator1()
 
@@ -318,6 +325,7 @@ class self_elt(data_elt):
       def loop2(iterator2, ca):
 
         def fail2(e):
+          rpki.log.error(traceback.format_exc())
           rpki.log.warn("Couldn't regenerate CRLs and manifests for CA %r, skipping: %s" % (ca, e))
           iterator2()
 
@@ -757,6 +765,7 @@ class route_origin_elt(data_elt):
     """
 
     def lose(e):
+      rpki.log.error(traceback.format_exc())
       rpki.log.warn("Could not update ROA %r, skipping: %s" % (self, e))
       callback()
       return
diff --git a/rpkid/rpki/rpki_engine.py b/rpkid/rpki/rpki_engine.py
index 9f19dfcd..33be7ca6 100644
--- a/rpkid/rpki/rpki_engine.py
+++ b/rpkid/rpki/rpki_engine.py
@@ -348,6 +348,7 @@ class ca_obj(rpki.sql.sql_persistent):
     """
 
     def fail(e):
+      rpki.log.error(traceback.format_exc())
       rpki.log.warn("Could not delete CA %r, skipping: %s" % (self, e))
       callback()
 
diff --git a/rpkid/testbed.2.yaml b/rpkid/testbed.2.yaml
index 36b5c17d..76f8d44d 100644
--- a/rpkid/testbed.2.yaml
+++ b/rpkid/testbed.2.yaml
@@ -45,6 +45,7 @@ kids:
         asn:  64534-64540
       - name: Harry
         asn:  666-677
+    #hosted_by: R1
   - name: R4
     kids:
       - name: Ilse
diff --git a/rpkid/testbed.py b/rpkid/testbed.py
index ca19b9d4..8c0ca96d 100644
--- a/rpkid/testbed.py
+++ b/rpkid/testbed.py
@@ -157,8 +157,6 @@ def main():
   rpki.log.init(testbed_name)
   rpki.log.info("Starting")
 
-  signal.signal(signal.SIGALRM, wakeup)
-
   pubd_process = None
   rootd_process = None
   rsyncd_process = None
@@ -224,12 +222,12 @@ def main():
     for a in db.engines:
       a.run_daemons()
 
-    rpki.log.info("Sleeping %d seconds while daemons start up" % startup_delay)
-    time.sleep(startup_delay)
-
     # From this point on we'll be running event-driven, so the rest of
     # the code until final exit is all closures.
 
+    def start():
+      rpki.async.iterator(db.engines, create_rpki_objects, created_rpki_objects)
+
     def create_rpki_objects(iterator, a):
       a.create_rpki_objects(iterator)
 
@@ -279,8 +277,8 @@ def main():
       # Loop until we run out of control YAML
       yaml_loop()
 
-    rpki.async.iterator(db.engines, create_rpki_objects, created_rpki_objects)
-
+    rpki.log.info("Sleeping %d seconds while daemons start up" % startup_delay)
+    rpki.async.timer(start).set(rpki.sundial.timedelta(seconds = startup_delay))
     rpki.async.event_loop()
 
     # At this point we have gone into event-driven code.
@@ -308,33 +306,27 @@ def main():
           pass
         proc.wait()
 
-def wakeup(signum, frame):
-  """Handler called when we receive a SIGALRM signal."""
-  rpki.log.info("Wakeup call received, continuing")
-
-def cmd_sleep(interval = None):
+def cmd_sleep(cb, interval):
   """
   Set an alarm, then wait for it to go off.
   """
-  if interval is None:
-    rpki.log.info("Pausing indefinitely, send a SIGALRM to wake me up")
-  else:
-    seconds = rpki.sundial.timedelta.parse(interval).convert_to_seconds()
-    rpki.log.info("Sleeping %s seconds" % seconds)
-    signal.alarm(seconds)
-  signal.pause()
-
-def cmd_shell(*cmd):
+  howlong = rpki.sundial.timedelta.parse(interval)
+  rpki.log.info("Sleeping %r" % howlong)
+  rpki.async.timer(cb).set(howlong)
+
+def cmd_shell(cb, *cmd):
   """
   Run a shell command.
   """
   cmd = " ".join(cmd)
   status = subprocess.call(cmd, shell = True)
   rpki.log.info("Shell command returned status %d" % status)
+  cb()
 
-def cmd_echo(*words):
+def cmd_echo(cb, *words):
   """Echo some text to the log."""
   rpki.log.note(" ".join(words))
+  cb()
 
 ## @var cmds
 # Dispatch table for commands embedded in delta sections
@@ -418,8 +410,7 @@ class allocation_db(list):
     def loop(iterator, d):
       if isinstance(d, str):
         c = d.split()
-        cmds[c[0]](*c[1:])
-        iterator()
+        cmds[c[0]](iterator, *c[1:])
       else:
         self.map[d["name"]].apply_delta(d, iterator)
 
@@ -924,22 +915,22 @@ class allocation(object):
         assert s.self_id == v.self_id
         s.repository_id = v.repository_id
 
-        rpki.log.info("Creating rpkid parent object for %s" % self.name)
+        rpki.log.info("Creating rpkid parent object for %s" % s.name)
 
-        if self.is_root():
-          rootd_cert = self.cross_certify(rootd_name + "-TA")
-          pdus.append(rpki.left_right.parent_elt.make_pdu(action = "create", tag = v.tag, self_id = self.self_id, bsc_id = self.bsc_id,
-                                                          repository_id = self.repository_id, sia_base = self.sia_base,
-                                                          bpki_cms_cert = rootd_cert, bpki_https_cert = rootd_cert, sender_name = self.name, recipient_name = "Walrus",
+        if s.is_root():
+          rootd_cert = s.cross_certify(rootd_name + "-TA")
+          pdus.append(rpki.left_right.parent_elt.make_pdu(action = "create", tag = v.tag, self_id = s.self_id, bsc_id = s.bsc_id,
+                                                          repository_id = s.repository_id, sia_base = s.sia_base,
+                                                          bpki_cms_cert = rootd_cert, bpki_https_cert = rootd_cert, sender_name = s.name, recipient_name = "Walrus",
                                                           peer_contact_uri = "https://localhost:%s/" % rootd_port))
         else:
-          parent_cms_cert = self.cross_certify(self.parent.name + "-SELF")
-          parent_https_cert = self.cross_certify(self.parent.name + "-TA")
-          pdus.append(rpki.left_right.parent_elt.make_pdu(action = "create", tag = v.tag, self_id = self.self_id, bsc_id = self.bsc_id,
-                                                          repository_id = self.repository_id, sia_base = self.sia_base,
+          parent_cms_cert = s.cross_certify(s.parent.name + "-SELF")
+          parent_https_cert = s.cross_certify(s.parent.name + "-TA")
+          pdus.append(rpki.left_right.parent_elt.make_pdu(action = "create", tag = v.tag, self_id = s.self_id, bsc_id = s.bsc_id,
+                                                          repository_id = s.repository_id, sia_base = s.sia_base,
                                                           bpki_cms_cert = parent_cms_cert, bpki_https_cert = parent_https_cert,
-                                                          sender_name = self.name, recipient_name = self.parent.name,
-                                                          peer_contact_uri = "https://localhost:%s/up-down/%s" % (self.parent.rpki_port, self.child_id)))
+                                                          sender_name = s.name, recipient_name = s.parent.name,
+                                                          peer_contact_uri = "https://localhost:%s/up-down/%s" % (s.parent.rpki_port, s.child_id)))
 
       self.call_rpkid(pdus, cb = got_parent_id)
 
@@ -963,7 +954,10 @@ class allocation(object):
                                                          bsc_id = s.bsc_id,
                                                          bpki_cert = s.cross_certify(k.name + ("-TA" if k.is_leaf() else "-SELF"))))
 
-      self.call_rpkid(pdus, cb = got_child_ids)
+      if pdus:
+        self.call_rpkid(pdus, cb = got_child_ids)
+      else:
+        got_child_ids(())
 
     def got_child_ids(vals):
 
@@ -994,7 +988,10 @@ class allocation(object):
                                                                 ipv4 = r.v4,
                                                                 ipv6 = r.v6))
 
-      self.call_rpkid(pdus, cb = got_route_origin_ids)
+      if pdus:
+        self.call_rpkid(pdus, cb = got_route_origin_ids)
+      else:
+        got_route_origin_ids(())
 
     def got_route_origin_ids(vals):
       
@@ -1047,7 +1044,7 @@ class allocation(object):
     rpki.log.info("Running cron for %s" % self.name)
 
     def done(result):
-      assert result == "OK"
+      assert result == "OK", 'Expected "OK" result from cronjob, got %r' % result
       cb()
 
     rpki.https.client(client_key   = self.irbe_key,