Merge lp:~stub/launchpad/replication into lp:launchpad/db-devel

[Stuart Bishop]

an alternative approach of stopping the appservers from getting blocked when they check replication lag under high load.

[Stuart Bishop (stub)]

The fix to Bug #504696 fails on production, despite passing testing locally and on staging and I'm unable to reproduce.

This branch implements an alternative approach of stopping the appservers from getting blocked when they check replication lag under high load.

The database patch, stored procedures and permissions have been applied to production in case we want to test this on edge while I'm unavailable.

[Aaron Bentley (abentley)]

review: Approve

=== added symlink 'daemons/_pythonpath.py'
=== target is u'../_pythonpath.py'
=== added file 'daemons/cache-database-replication-lag.py'
--- daemons/cache-database-replication-lag.py	1970-01-01 00:00:00 +0000
+++ daemons/cache-database-replication-lag.py	2010-02-15 07:48:22 +0000
@@ -0,0 +1,53 @@
+#!/usr/bin/python2.5
+#
+# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Calculate database replication lag and cache it."""
+
+__metaclass__ = type
+__all__ = []
+
+import _pythonpath
+
+import sys
+import time
+
+import psycopg2
+
+from canonical.database.sqlbase import connect, ISOLATION_LEVEL_AUTOCOMMIT
+from canonical.launchpad.scripts import db_options, logger
+from lp.scripts.helpers import LPOptionParser
+
+
+def main(args=None):
+    parser = LPOptionParser()
+    db_options(parser)
+    parser.add_option(
+        "-s", "--sleep", dest="sleep", type="int", default=5,
+        metavar="SECS", help="Wait SECS seconds between refreshes.")
+
+    (options, args) = parser.parse_args(args)
+    if len(args) != 0:
+        parser.error("Too many arguments.")
+
+    log = logger(options)
+
+    while True:
+        try:
+            con = connect(user="lagmon", isolation=ISOLATION_LEVEL_AUTOCOMMIT)
+            cur = con.cursor()
+            while True:
+                cur.execute("SELECT update_replication_lag_cache()")
+                if cur.fetchone()[0]:
+                    log.info("Updated.")
+                else:
+                    log.error("update_replication_lag_cache() failed.")
+                time.sleep(options.sleep)
+        except psycopg2.Error, x:
+            log.error("%s. Retrying.", str(x).strip())
+            time.sleep(options.sleep)
+
+
+if __name__ == '__main__':
+    sys.exit(main())

=== modified file 'database/replication/helpers.py'
--- database/replication/helpers.py	2010-01-22 06:25:48 +0000
+++ database/replication/helpers.py	2010-02-15 07:48:22 +0000
@@ -44,6 +44,7 @@
 LPMAIN_SEED = frozenset([
     ('public', 'person'),
     ('public', 'launchpaddatabaserevision'),
+    ('public', 'databasereplicationlag'),
     ('public', 'fticache'),
     ('public', 'nameblacklist'),
     ('public', 'openidconsumerassociation'),

=== modified file 'database/schema/comments.sql'
--- database/schema/comments.sql	2010-02-03 15:55:01 +0000
+++ database/schema/comments.sql	2010-02-15 07:48:22 +0000
@@ -2352,3 +2352,7 @@
 COMMENT ON COLUMN SourcePackageFormatSelection.distroseries IS 'Refers to the distroseries in question.';
 COMMENT ON COLUMN SourcePackageFormatSelection.format IS 'The SourcePackageFormat to allow.';
 
+COMMENT ON TABLE DatabaseReplicationLag IS 'A cached snapshot of database replication lag between our master Slony node and its slaves.';
+COMMENT ON COLUMN DatabaseReplicationLag.node IS 'The Slony node number identifying the slave database.';
+COMMENT ON COLUMN DatabaseReplicationLag.lag IS 'lag time.';
+COMMENT ON COLUMN DatabaseReplicationLag.updated IS 'When this value was updated.';

=== added file 'database/schema/patch-2207-28-1.sql'
--- database/schema/patch-2207-28-1.sql	1970-01-01 00:00:00 +0000
+++ database/schema/patch-2207-28-1.sql	2010-02-15 07:48:22 +0000
@@ -0,0 +1,9 @@
+SET client_min_messages = ERROR;
+
+CREATE TABLE DatabaseReplicationLag (
+    node integer PRIMARY KEY,
+    lag interval NOT NULL,
+    updated timestamp without time zone
+        DEFAULT (CURRENT_TIMESTAMP AT TIME ZONE 'UTC'));
+
+INSERT INTO LaunchpadDatabaseRevision VALUES (2207, 28, 1);

=== modified file 'database/schema/security.cfg'
--- database/schema/security.cfg	2010-02-12 15:44:24 +0000
+++ database/schema/security.cfg	2010-02-15 07:48:22 +0000
@@ -166,6 +166,7 @@
 public.cvereference                     = SELECT, INSERT
 public.cve                              = SELECT, INSERT, UPDATE
 public.customlanguagecode               = SELECT, INSERT, UPDATE, DELETE
+public.databasereplicationlag           = SELECT
 public.diff                             = SELECT, INSERT, UPDATE
 public.distributionbounty               = SELECT, INSERT, UPDATE
 public.distributionmirror               = SELECT, INSERT, UPDATE, DELETE
@@ -1898,6 +1899,11 @@
 public.job                              = SELECT, UPDATE, DELETE
 public.bugjob                           = SELECT, DELETE
 
+[lagmon]
+# cache-database-replication-lag.py
+type=user
+public.update_replication_lag_cache()   = EXECUTE
+
 [process-apport-blobs]
 type=user
 groups=script,read

=== modified file 'database/schema/trusted.sql'
--- database/schema/trusted.sql	2010-02-02 10:57:26 +0000
+++ database/schema/trusted.sql	2010-02-15 07:48:22 +0000
@@ -94,6 +94,28 @@
 'Returns the lag time of the lpmain replication set to the given node, or NULL if not a replicated installation. The node id parameter can be obtained by calling getlocalnodeid() on the relevant database. This function only returns meaningful results on the lpmain replication set master.';
 
 
+CREATE OR REPLACE FUNCTION update_replication_lag_cache() RETURNS boolean
+LANGUAGE plpgsql VOLATILE SECURITY DEFINER AS
+$$
+    BEGIN
+        DELETE FROM DatabaseReplicationLag;
+        INSERT INTO DatabaseReplicationLag (node, lag)
+            SELECT st_received, st_lag_time FROM _sl.sl_status
+            WHERE st_origin = _sl.getlocalnodeid('_sl');
+        RETURN TRUE;
+    -- Slony-I not installed here - non-replicated setup.
+    EXCEPTION
+        WHEN invalid_schema_name THEN
+            RETURN FALSE;
+        WHEN undefined_table THEN
+            RETURN FALSE;
+    END;
+$$;
+
+COMMENT ON FUNCTION update_replication_lag_cache() IS
+'Updates the DatabaseReplicationLag materialized view.';
+
+
 CREATE OR REPLACE FUNCTION getlocalnodeid() RETURNS integer
 LANGUAGE plpgsql STABLE SECURITY DEFINER AS
 $$

=== modified file 'lib/canonical/launchpad/webapp/dbpolicy.py'
--- lib/canonical/launchpad/webapp/dbpolicy.py	2010-01-20 22:09:26 +0000
+++ lib/canonical/launchpad/webapp/dbpolicy.py	2010-02-15 07:48:22 +0000
@@ -293,26 +293,18 @@
 
         # sl_status gives meaningful results only on the origin node.
         master_store = self.getStore(MAIN_STORE, MASTER_FLAVOR)
-        # If it takes more than (by default) 0.25 seconds to query the
-        # replication lag, assume we are lagged. Normally the query
-        # takes <20ms. This can happen during heavy updates, as the
-        # Slony-I tables can get slow with lots of events. We use a
-        # SAVEPOINT to conveniently reset the statement timeout.
-        master_store.execute("""
-            SAVEPOINT lag_check; SET LOCAL statement_timeout TO %d
-            """ % config.launchpad.lag_check_timeout)
-        try:
-            try:
-                return master_store.execute(
-                    "SELECT replication_lag(%d)" % slave_node_id).get_one()[0]
-            except TimeoutError:
-                logging.warn(
-                    'Gave up querying slave lag after %d ms',
-                    (config.launchpad.lag_check_timeout))
-                return timedelta(days=999) # A long, long time.
-        finally:
-            master_store.execute("ROLLBACK TO lag_check")
 
+        # Retrieve the cached lag.
+        lag = master_store.execute("""
+            SELECT lag + (CURRENT_TIMESTAMP AT TIME ZONE 'UTC' - updated)
+            FROM DatabaseReplicationLag WHERE node=%d
+            """ % slave_node_id).get_one()
+        if lag is None:
+            logging.error(
+                "No data in DatabaseReplicationLag for node %d"
+                % slave_node_id)
+            return timedelta(days=999) # A long, long time.
+        return lag[0]
 
 
 def WebServiceDatabasePolicyFactory(request):