Merge lp:~cmiller/desktopcouch/replicate-to-u1 into lp:desktopcouch

[Chad Miller]

Include ability to un-pair machines; for non-cloud machines, this adds a new attribute to the record, which the replicator reconciles with the other end when it sees it next.

Make the replication manager part of the desktopcouch-service program.

Make services a module with a standard interface, so anyone can add more later; this may need more work; a single instance does not make a good API.

[Joshua Blount (jblount)]

Looks handsome

review: Approve

[dobey (dobey)]

Josh says it's handsome, so it must be OK.

review: Approve

[Chad Miller (cmiller)]

I'm restructuring at Rodney's request.

review: Needs Fixing

[dobey (dobey)]

review: Needs Resubmitting

[Chad Miller (cmiller)]

Now incorporates aquarius' approved branch.

[Chad Miller (cmiller)]

review: Abstain

[Stuart Langridge (sil)]

Looks good, I think.

review: Approve

[Chad Miller (cmiller)]

desktopcouch.contacts.tests.test_contactspicker
  TestContactsPicker
    test_can_contruct_contactspicker ... Apache CouchDB has started, time to relax.
Desktop CouchDB is not running; starting it. ...waiting for couchdb to start...
...waiting for couchdb to start...
Browse your desktop CouchDB at file:///tmp/tmpVMvmKA/xdg_data/desktop-couch/couchdb.html
                                  [OK]
desktopcouch.contacts.tests.test_create
  TestCreate
    test_create_many_contacts ... [OK]
    test_head_or_tails ... [OK]
    test_random_bools ... [OK]
desktopcouch.contacts.tests.test_record
  TestContactRecord
    test_contact_record ... [OK]
                                            [ERROR]
desktopcouch.pair.tests.test_network_io
  TestNetworkIO
    test_successful_lifespan ... [OK]
desktopcouch.records.tests.test_couchgrid
  TestCouchGrid
    test_all_from_database ... [OK]
    test_constructor_guarded ... [OK]
    test_new_rows_with_headings ... [OK]
    test_no_headings_or_stored_records ... [OK]
    test_optional_args_no_stored_records ... [OK]
    test_optional_record_type_arg ... [OK]
    test_programatically_add_row ... [OK]
    test_single_col_from_database ... [OK]
desktopcouch.records.tests.test_field_registry
  TestFieldMapping
    test_mergeable_list_field_mapping ... [OK]
    test_mergeable_list_field_mapping_empty_field ... [OK]
    test_simple_field_mapping ... [OK]
  TestTransformer
    test_from_app ... [OK]
    test_to_app ... [OK]
desktopcouch.records.tests.test_record
  TestRecordFactory
    test_build ... [OK]
  TestRecords
    test_application_annotations ... [OK]
    test_dictionary_access_to_mergeable_list ... [OK]
    test_get ... [OK]
    test_get_item ... [OK]
    test_keys ... [OK]
    test_list ... [OK]
    test_loads_dict_multi_subdict ... [OK]
    test_loads_dict_subdict ... [OK]
    test_mergeable_list_append ... [OK]
    test_mergeable_list_append_record_dict ... [OK]
    test_mergeable_list_del ... ...

=== modified file 'bin/desktopcouch-pair'
--- bin/desktopcouch-pair	2009-08-26 18:01:29 +0000
+++ bin/desktopcouch-pair	2009-09-03 21:26:51 +0000
@@ -467,7 +467,7 @@
                 "consumer_secret": "",
             })
             listening_hosts.append(None, ["Ubuntu One", "The Ubuntu One cloud service", 
-               "ubuntuone.com", 5984, True])
+               "couchdb.one.ubuntu.com", 5984, True])
             CLOUD_SERVICES["Ubuntu One"] = oauth_data
 
         self.inviting = None  # pylint: disable-msg=W0201

=== modified file 'bin/desktopcouch-paired-replication-manager'
--- bin/desktopcouch-paired-replication-manager	2009-08-26 19:04:39 +0000
+++ bin/desktopcouch-paired-replication-manager	2009-09-03 21:26:51 +0000
@@ -37,6 +37,21 @@
 
 already_replicating = False
 
+
+def db_prefix_for_statically_addressed_replicators(addr, port):
+    addrport = addr + "_" + str(port)
+    module_name = addrport.replace(".", "_")
+    try:
+        logging.debug("Looking up prefix %r in mod %s", addrport, module_name)
+        mod = __import__("desktopcouch.replication_hosts", fromlist=[module_name])
+        return getattr(mod, module_name).db_name_prefix
+    except ImportError, e:
+        logging.info("Not changing remote db name.  %s", e)
+        return ""
+    except Exception, e:
+        logging.exception("Not changing remote db name.")
+        return ""
+
 class ReplicatorThread(threading.Thread):
     def __init__(self, local_port):
         log.debug("starting up replication thread")
@@ -48,15 +63,34 @@
         already_replicating = True  #   just trying to be polite.
         try:
             for uuid, addr, port in dbus_io.get_seen_paired_hosts():
-                log.debug("host %s is seen; want to replicate to it", uuid)
+                log.debug("want to replipush to discovered host %r @ %s",
+                        uuid, addr)
                 for db_name in couchdb_io.get_database_names_replicatable():
                     couchdb_io.replicate(db_name, db_name,
                             target_host=addr, target_port=port, 
                             source_port=self.local_port)
-
-
-            # TODO: get static addressed paired hosts and replicate to
-            # those too.
+            
+            for uuid, addr, port, to_pull, to_push in \
+                    couchdb_io.get_static_paired_hosts():
+                
+                if to_pull:
+                    remote_db_prefix = db_prefix_for_statically_addressed_replicators(addr, port)
+                    for db_name in couchdb_io.get_database_names_replicatable():
+                        remote_db_name = str(remote_db_prefix)+db_name
+                        log.debug("want to replipush %r to static host %r @ %s",
+                                remote_db_name, uuid, addr)
+                        couchdb_io.replicate(db_name, remote_db_name,
+                                target_host=addr, target_port=port, 
+                                source_port=self.local_port, target_ssl=True)
+                if to_push:
+                    for db_name in couchdb_io.get_database_names_replicatable(addr,
+                            port):
+                        remote_db_name = str(remote_db_prefix)+db_name
+                        log.debug("want to replipull %r from static host %r @ %s",
+                                remote_db_name, uuid, addr)
+                        couchdb_io.replicate(remote_db_name, db_name,
+                                source_host=addr, source_port=port, 
+                                target_port=self.local_port, source_ssl=True)
 
         finally:
             already_replicating = False
@@ -70,6 +104,7 @@
 
     r = ReplicatorThread(local_port)
     r.start()
+    return r
 
 def main(args):
     log_directory = os.path.join(xdg.BaseDirectory.xdg_cache_home,
@@ -87,6 +122,8 @@
     logging.getLogger('').addHandler(rotating_log)
     logging.getLogger('').setLevel(logging.DEBUG)
 
+    print "Logging to", os.path.join(log_directory, "desktop-couch-replication.log")
+
     try:
         log.info("Starting.")
 
@@ -117,6 +154,10 @@
                 b.unpublish()
 
     finally:
+        try:
+            t.stop()
+        except:
+            pass
         log.info("Quitting.")
 
 if __name__ == "__main__":

=== modified file 'desktopcouch/pair/couchdb_pairing/couchdb_io.py'
--- desktopcouch/pair/couchdb_pairing/couchdb_io.py	2009-08-26 19:04:39 +0000
+++ desktopcouch/pair/couchdb_pairing/couchdb_io.py	2009-09-03 21:26:51 +0000
@@ -21,6 +21,7 @@
 
 from desktopcouch import find_port as desktopcouch_find_port
 from desktopcouch.records import server
+import socket
 
 RECTYPE_BASE = "http://www.freedesktop.org/wiki/Specifications/desktopcouch/"
 PAIRED_SERVER_RECORD_TYPE = RECTYPE_BASE + "paired_server"
@@ -30,13 +31,40 @@
     port = desktopcouch_find_port()  # make sure d-c is running.
     return server.CouchDatabase(name, create=create)
 
-def get_database_names_replicatable():
+def get_static_paired_hosts():
+    db = _get_db("management")
+    results = db.get_records(create_view=True)
+    found = dict()
+    for row in results[PAIRED_SERVER_RECORD_TYPE]:
+        try:
+            if row.value["server"] != "":
+                uuid = row.value["pairing_identifier"]
+                to_push = row.value.get("push_to_server", True)
+                to_pull = row.value.get("pull_from_server", False)
+                addr, port = row.value["server"].split(":") # What about IPv6?
+                found[(addr, int(port))] = uuid, to_pull, to_push
+        except KeyError:
+            pass
+    unique_hosts = [(v1, h1, h2, v2, v3) for 
+            (h1, h2), (v1, v2, v3) in found.items()]
+    logging.debug("static pairings are %s", unique_hosts)
+    return unique_hosts
+
+def get_database_names_replicatable(host=None, port=None):
     """Find a list of local databases, minus dbs that we do not want to
     replicate (explicitly or implicitly)."""
 
-    port = int(desktopcouch_find_port())
-    couchdb_server = server.Server("http://localhost:%(port)d/" % locals())
-    all = set([db_name for db_name in couchdb_server])
+    if host is None:
+        host = "localhost"
+    if port is None:
+        port = int(desktopcouch_find_port())
+
+    try:
+        couchdb_server = server.Server("http://%(host)s:%(port)d/" % locals())
+        all = set([db_name for db_name in couchdb_server])
+    except socket.error, e:
+        logging.error("Can't get list of databases from %s", couchdb_server)
+        return set()
 
     excluded = set()
     excluded.add("management")
@@ -82,51 +110,52 @@
     logging.debug("found %d %s records", len(values), key)
     return values
 
-def create_remote_database(dst_host, dst_port, dst_name):
+def create_database(dst_host, dst_port, dst_name):
     dst_url = u"http://%(dst_host)s:%(dst_port)d/" % locals()
     return server.CouchDatabase(dst_name, dst_url, create=True)
 
 def replicate(source_database, target_database, target_host=None,
-        target_port=None, source_host=None, source_port=None):
+        target_port=None, source_host=None, source_port=None,
+        source_ssl=False, target_ssl=False):
     """This replication is instant and blocking, and does not persist. """
 
-    data = {}
-
+    source_protocol = "https" if source_ssl else "http"
+    target_protocol = "https" if target_ssl else "http"
     if source_host:
         if source_port is None:
-            source = "http://%(source_host)s/%(source_database)s" % locals()
+            source = "%(source_protocol)s://%(source_host)s/%(source_database)s" % locals()
         else:
-            source = "http://%(source_host)s:%(source_port)d/%(source_database)s" % locals()
+            source = "%(source_protocol)s://%(source_host)s:%(source_port)d/%(source_database)s" % locals()
     else:
         source = source_database
 
     if target_host:
         if target_port is None:
-            target = "http://%(target_host)s/%(target_database)s" % locals()
+            target = "%(target_protocol)s://%(target_host)s/%(target_database)s" % locals()
         else:
-            target = "http://%(target_host)s:%(target_port)d/%(target_database)s" % locals()
+            target = "%(target_protocol)s://%(target_host)s:%(target_port)d/%(target_database)s" % locals()
     else:
         target = target_database
             
     record = dict(source=source, target=target)
     try:
-        if target_host:
-            # Remote databases must exist before replicating to them.
-            create_remote_database(target_host, target_port, target_database)
-
+        url = None  # so logging works in exception handler
+        port = int(desktopcouch_find_port())
         # TODO: Get admin username and password from keyring.  Populate URL.
-
-        url = None  # so logging works in exception handler
-        port = int(desktopcouch_find_port())
         url = "http://localhost:%d/" % (port,)
 
+        if target_host:
+            # Target databases must exist before replicating to them.
+            logging.debug("creating %r %s:%d", target_database, target_host, target_port)
+            create_database(target_host, target_port, target_database)
+
+
         ### All until python-couchdb gets a Server.replicate() function
         import couchdb
+        logging.debug("asking %r to send %s to %s", url, source, target)
         server = couchdb.client.Server(url)
         resp, data = server.resource.post(path='/_replicate', content=record)
         logging.debug("replicate result:  %r  %r", resp, data)
         ###
-        logging.debug("#############################")
     except:
         logging.exception("can't talk to couchdb.  %r <== %r", url, record)
-        raise

=== added directory 'desktopcouch/replication_hosts'
=== added file 'desktopcouch/replication_hosts/__init__.py'
=== added file 'desktopcouch/replication_hosts/couchdb_one_ubuntu_com.py'
--- desktopcouch/replication_hosts/couchdb_one_ubuntu_com.py	1970-01-01 00:00:00 +0000
+++ desktopcouch/replication_hosts/couchdb_one_ubuntu_com.py	2009-09-03 21:26:51 +0000
@@ -0,0 +1,86 @@
+import hashlib
+from oauth import oauth
+import logging
+import httplib2
+import simplejson
+import gnomekeyring
+
+
+def whoami_user_id():
+    def get_oauth_token(consumer):
+        """Get the token from the keyring"""
+        import gobject
+        gobject.set_application_name("desktopcouch replication to Ubuntu One")
+        items = gnomekeyring.find_items_sync(
+            gnomekeyring.ITEM_GENERIC_SECRET,
+            {'ubuntuone-realm': "https://one.ubuntu.com",
+             'oauth-consumer-key': consumer.key})
+        if len(items):
+            return oauth.OAuthToken.from_string(items[0].secret)
+
+    def get_oauth_request_header(consumer, access_token, http_url):
+        """Get an oauth request header given the token and the url"""
+        signature_method = oauth.OAuthSignatureMethod_HMAC_SHA1()
+        oauth_request = oauth.OAuthRequest.from_consumer_and_token(
+            http_url=http_url,
+            http_method="GET",
+            oauth_consumer=consumer,
+            token=access_token)
+        oauth_request.sign_request(signature_method, consumer, access_token)
+        return oauth_request.to_header()
+
+    url = "https://one.ubuntu.com/api/account/"
+    consumer = oauth.OAuthConsumer("ubuntuone", "hammertime")
+    try:
+        access_token = get_oauth_token(consumer)
+    except gnomekeyring.NoKeyringDaemonError:
+        logging.info("No keyring daemon is running for this session.")
+        return None
+    if not access_token:
+        logging.info("Could not get access token from keyring")
+        return None
+    oauth_header = get_oauth_request_header(consumer, access_token, url)
+    client = httplib2.Http()
+    resp, content = client.request(url, "GET", headers=oauth_header)
+    if resp['status'] == "200":
+        try:
+            userinfo = simplejson.loads(content)
+            return userinfo["id"]
+        except:
+            logging.error("Couldn't find id in response %r", content)
+            return None
+    else:
+        logging.error("Couldn't talk to %r.  Got HTTP %s", url, resp['status'])
+        return None
+
+
+class PrefixGetter():
+    def __init__(self):
+        self.str = None
+
+    def __str__(self):
+        if self.str is not None:
+            return self.str
+
+        user_id = whoami_user_id()
+        if user_id is None:
+            raise ValueError
+
+        hasher = hashlib.md5()
+        hasher.update(str(user_id))
+        hashed_id_as_hex = hasher.hexdigest()
+
+        prefix = "u/%s/%s/%d/" % (
+                hashed_id_as_hex[0:3],
+                hashed_id_as_hex[3:6],
+                user_id)
+        self.str = prefix
+        return prefix
+
+
+# Access to this as a string fires off functions.
+db_name_prefix = PrefixGetter()
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.DEBUG, format="%(message)s")
+    print str(db_name_prefix)