From 2455458888ee9bd20ed1e16fac1905d02f89615a Mon Sep 17 00:00:00 2001
From: Mark Reynolds <mreynolds@redhat.com>
Date: Mon, 26 Jan 2015 16:11:18 -0500
Subject: [PATCH] Ticket 48000 - replica agreement pause/resume should have a
 short sleep

Bug Description:  There is a small window between disabling a replica agreement and it actually
                  stopping the agreement thread. This can allow one more replication session
                  to slip through, which can cause issues with lib389 tests. Replication logging
                  can also widen this "window".

Fix Description:  Add a short sleep after pausing/resuming a replication agreement.

https://fedorahosted.org/389/ticket/48000

Reviewed by: ?
---
 lib389/agreement.py | 218 ++++++++++++++++++++++++++--------------------------
 1 file changed, 108 insertions(+), 110 deletions(-)

diff --git a/lib389/agreement.py b/lib389/agreement.py
index 561e499..141f6df 100644
--- a/lib389/agreement.py
+++ b/lib389/agreement.py
@@ -6,6 +6,7 @@ Created on Dec 5, 2013
 
 import ldap
 import re
+import time
 
 from lib389._constants import *
 from lib389._entry import FormatDict
@@ -13,12 +14,13 @@ from lib389.utils import normalizeDN
 from lib389 import Entry, DirSrv, NoSuchEntryError, InvalidArgumentError
 from lib389.properties import *
 
+
 class Agreement(object):
     ALWAYS = '0000-2359 0123456'
     NEVER = '2358-2359 0'
-    
+
     proxied_methods = 'search_s getEntry'.split()
-    
+
     def __init__(self, conn):
         """@param conn - a DirSrv instance"""
         self.conn = conn
@@ -27,16 +29,14 @@ class Agreement(object):
     def __getattr__(self, name):
         if name in Agreement.proxied_methods:
             return DirSrv.__getattr__(self.conn, name)
-        
 
-    
     def status(self, agreement_dn):
         """Return a formatted string with the replica status. Looking like:
             Status for meTo_localhost.localdomain:50389 agmt localhost.localdomain:50389
             Update in progress: TRUE
             Last Update Start: 20131121132756Z
             Last Update End: 0
-            Num. Changes Sent: 1:10/0 
+            Num. Changes Sent: 1:10/0
             Num. changes Skipped: None
             Last update Status: 0 Replica acquired successfully: Incremental update started
             Init in progress: None
@@ -88,18 +88,18 @@ class Agreement(object):
             HH [0..23]
             MM [0..59]
             DAYS [0-6]{1,7}
-            
+
             @param interval - interval in the format 'HHMM-HHMM D+' (D is day number [0-6])
-            
+
             @return None
-            
+
             @raise ValueError - if the inteval is illegal
         '''
         c = re.compile(re.compile('^([0-9][0-9])([0-9][0-9])-([0-9][0-9])([0-9][0-9]) ([0-6]{1,7})$'))
         if not c.match(interval):
             raise ValueError("Bad schedule format %r" % interval)
         schedule = c.split(interval, c.groups)
-        
+
         # check the hours
         hour = int(schedule[1])
         if ((hour < 0) or (hour > 23)):
@@ -109,7 +109,7 @@ class Agreement(object):
             raise ValueError("Bad schedule format %r: illegal hour %d" % (interval, hour))
         if int(schedule[1]) > int(schedule[3]):
             raise ValueError("Bad schedule (start HOUR larger than end HOUR) %r: illegal hour %d" % (interval, int(schedule[1])))
-        
+
         # check the minutes
         minute = int(schedule[2])
         if ((minute < 0) or (minute > 59)):
@@ -118,45 +118,42 @@ class Agreement(object):
         if ((minute < 0) or (minute > 59)):
             raise ValueError("Bad schedule format %r: illegal minute %d" % (interval, minute))
 
-
-
-        
     def schedule(self, agmtdn=None, interval=ALWAYS):
         """Schedule the replication agreement
             @param agmtdn - DN of the replica agreement
-            @param interval - in the form 
+            @param interval - in the form
                     - Agreement.ALWAYS
                     - Agreement.NEVER
                     - or 'HHMM-HHMM D+' With D=[0123456]+
-                    
+
             @return - None
-            
+
             @raise ValueError - if interval is not valid
                    ldap.NO_SUCH_OBJECT - if agmtdn does not exist
         """
         if not agmtdn:
             raise InvalidArgumentError("agreement DN is missing")
-        
+
         # check the validity of the interval
         if interval != Agreement.ALWAYS and interval != Agreement.NEVER:
             self._check_interval(interval)
-        
+
         # Check if the replica agreement exists
         try:
             self.conn.getEntry(agmtdn, ldap.SCOPE_BASE)
         except ldap.NO_SUCH_OBJECT:
             raise
-            
+
         # update it
         self.log.info("Schedule replication agreement %s" % agmtdn)
         mod = [(
             ldap.MOD_REPLACE, 'nsds5replicaupdateschedule', [interval])]
         self.conn.modify_s(agmtdn, mod)
-        
+
     def getProperties(self, agmnt_dn=None, properties=None):
         '''
-            returns a dictionary of the requested properties. If properties is missing, 
-            it returns all the properties. 
+            returns a dictionary of the requested properties. If properties is missing,
+            it returns all the properties.
             @param agmtdn - is the replica agreement DN
             @param properties - is the list of properties name
             Supported properties are
@@ -178,17 +175,17 @@ class Agreement(object):
                 RA_CHANGES
 
             @return - returns a dictionary of the properties
-            
+
             @raise ValueError - if invalid property name
                    ldap.NO_SUCH_OBJECT - if agmtdn does not exist
                    InvalidArgumentError - missing mandatory argument
-            
-            
-        ''' 
-        
+
+
+        '''
+
         if not agmnt_dn:
             raise InvalidArgumentError("agmtdn is a mandatory argument")
-        
+
         #
         # prepare the attribute list to retrieve from the RA
         # if properties is None, all RA attributes are retrieved
@@ -200,32 +197,32 @@ class Agreement(object):
                 if not prop_attr:
                     raise ValueError("Improper property name: %s ", prop_name)
                 attrs.append(prop_attr)
-            
-        filt = "(objectclass=*)" 
+
+        filt = "(objectclass=*)"
         result = {}
         try:
             entry = self.conn.getEntry(agmnt_dn, ldap.SCOPE_BASE, filt, attrs)
-                
+
             # Build the result from the returned attributes
             for attr in entry.getAttrs():
                 # given an attribute name retrieve the property name
-                props = [ k for k, v in RA_PROPNAME_TO_ATTRNAME.iteritems() if v.lower() == attr.lower() ]
-                
+                props = [k for k, v in RA_PROPNAME_TO_ATTRNAME.iteritems() if v.lower() == attr.lower()]
+
                 # If this attribute is present in the RA properties, adds it to result
-                if len(props) > 0: 
+                if len(props) > 0:
                     result[props[0]] = entry.getValues(attr)
 
         except ldap.NO_SUCH_OBJECT:
             raise
-        
+
         return result
-        
+
     def setProperties(self, suffix=None, agmnt_dn=None, agmnt_entry=None, properties=None):
         '''
             Set the properties of the agreement. If an 'agmnt_entry' (Entry) is provided, it updates the entry, else
             it updates the entry on the server. If the 'agmnt_dn' is provided it retrieves the entry using it, else
             it retrieve the agreement using the 'suffix'.
-            
+
             @param suffix : suffix stored in that agreement (online update)
             @param agmnt_dn: DN of the agreement (online update)
             @param agmnt_entry: Entry of a agreement (offline update)
@@ -248,20 +245,20 @@ class Agreement(object):
                 RA_TIMEOUT
                 RA_CHANGES
 
-            
+
             @return None
-            
+
             @raise ValueError: if unknown properties
                     ValueError: if invalid agreement_entry
-                    ValueError: if agmnt_dn or suffix are not associated to a replica 
+                    ValueError: if agmnt_dn or suffix are not associated to a replica
                     InvalidArgumentError: If missing mandatory parameter
-            
-            
-        ''' 
+
+
+        '''
         # No properties provided
         if len(properties) == 0:
             return
-        
+
         # check that the given properties are valid
         for prop in properties:
             # skip the prefix to add/del value
@@ -269,33 +266,33 @@ class Agreement(object):
                 raise ValueError("unknown property: %s" % prop)
             else:
                 self.log.debug("setProperties: %s:%s" % (prop, properties[prop]))
-                
+
         # At least we need to have suffix/agmnt_dn/agmnt_entry
         if not suffix and not agmnt_dn and not agmnt_entry:
             raise InvalidArgumentError("suffix and agmnt_dn and agmnt_entry are missing")
-        
+
         # TODO
         if suffix:
             raise NotImplemented
-        
+
                 # the caller provides a set of properties to set into a replica entry
         if agmnt_entry:
             if not isinstance(agmnt_entry, Entry):
                 raise ValueError("invalid instance of the agmnt_entry")
-                      
+
             # that is fine, now set the values
             for prop in properties:
                 val = rawProperty(prop)
-                
+
                 # for Entry update it is a replace
                 agmnt_entry.update({RA_PROPNAME_TO_ATTRNAME[val]: properties[prop]})
-                
-            return   
-        
+
+            return
+
         # for each provided property build the mod
         mod = []
         for prop in properties:
-            
+
             # retrieve/check the property name
             # and if the value needs to be added/deleted/replaced
             if prop.startswith('+'):
@@ -307,21 +304,20 @@ class Agreement(object):
             else:
                 mod_type = ldap.MOD_REPLACE
                 prop_name = prop
-            
+
             attr = RA_PROPNAME_TO_ATTRNAME[prop_name]
             if not attr:
                 raise ValueError("invalid property name %s" % prop_name)
-            
+
             # Now do the value checking we can do
             if prop_name == RA_SCHEDULE:
                 self._check_interval(properties[prop])
-            
+
             mod.append((mod_type, attr, properties[prop]))
-        
+
         # Now time to run the effective modify
         self.conn.modify_s(agmnt_dn, mod)
-        
-    
+
     def list(self, suffix=None, consumer_host=None, consumer_port=None, agmtdn=None):
         '''
             Returns the search result of the replica agreement(s) under the replica (replicaRoot is 'suffix').
@@ -330,28 +326,28 @@ class Agreement(object):
             'consumer_host' and 'consumer_port' are either not specified or specified both.
 
             If 'agmtdn' is specified, it returns the search result entry of that replication agreement.
-            else if consumer host/port are specified it returns the replica agreements toward 
+            else if consumer host/port are specified it returns the replica agreements toward
             that consumer host:port.
-            Finally if neither 'agmtdn' nor 'consumser host/port' are specifies it returns 
+            Finally if neither 'agmtdn' nor 'consumser host/port' are specifies it returns
             all the replica agreements under the replica (replicaRoot is 'suffix').
 
-            @param - suffix is the suffix targeted by the total update 
+            @param - suffix is the suffix targeted by the total update
             @param - consumer_host hostname of the consumer
             @param - consumer_port port of the consumer
             @param - agmtdn DN of the replica agreement
- 
+
             @return - search result of the replica agreements
- 
+
             @raise - InvalidArgument: if missing mandatory argument (agmtdn or suffix, then host and port)
                    - ValueError - if some properties are not valid
                    - NoSuchEntryError - If no replica defined for the suffix
         '''
         if not suffix and not agmtdn:
             raise InvalidArgumentError("suffix or agmtdn are required")
-        
+
         if (consumer_host and not consumer_port) or (not consumer_host and consumer_port):
             raise InvalidArgumentError("consumer host/port are required together")
-        
+
         if agmtdn:
             # easy case, just return the RA
             filt = "objectclass=*"
@@ -362,18 +358,17 @@ class Agreement(object):
             if not replica_entries:
                 raise NoSuchEntryError("Error: no replica set up for suffix " + suffix)
             replica_entry = replica_entries[0]
-            
+
             # Now returns the replica agreement for that suffix that replicates to
             # consumer host/port
             if consumer_host and consumer_port:
                 filt = "(&(objectclass=%s)(%s=%s)(%s=%d))" % (RA_OBJECTCLASS_VALUE,
-                                                              RA_PROPNAME_TO_ATTRNAME[RA_CONSUMER_HOST], consumer_host, 
+                                                              RA_PROPNAME_TO_ATTRNAME[RA_CONSUMER_HOST], consumer_host,
                                                               RA_PROPNAME_TO_ATTRNAME[RA_CONSUMER_PORT], consumer_port)
             else:
                 filt = "(objectclass=%s)" % RA_OBJECTCLASS_VALUE
             return self.conn.search_s(replica_entry.dn, ldap.SCOPE_ONELEVEL, filt)
-        
-        
+
     def create(self, suffix=None, host=None, port=None, properties=None):
         """Create (and return) a replication agreement from self to consumer.
             - self is the supplier,
@@ -382,42 +377,42 @@ class Agreement(object):
                     * a DirSrv object if chaining
                     * an object with attributes: host, port, sslport, __str__
             @param suffix    - eg. 'dc=babel,dc=it'
-            @param properties      - further properties dict. 
+            @param properties      - further properties dict.
             Support properties
                 RA_NAME
-                RA_SUFFIX               
-                RA_BINDDN               
-                RA_BINDPW               
-                RA_METHOD               
-                RA_DESCRIPTION          
-                RA_SCHEDULE             
-                RA_TRANSPORT_PROT       
-                RA_FRAC_EXCLUDE         
-                RA_FRAC_EXCLUDE_TOTAL_UPDATE 
-                RA_FRAC_STRIP           
-                RA_CONSUMER_PORT        
-                RA_CONSUMER_HOST        
-                RA_CONSUMER_TOTAL_INIT  
-                RA_TIMEOUT              
-                RA_CHANGES             
-
-                    
+                RA_SUFFIX
+                RA_BINDDN
+                RA_BINDPW
+                RA_METHOD
+                RA_DESCRIPTION
+                RA_SCHEDULE
+                RA_TRANSPORT_PROT
+                RA_FRAC_EXCLUDE
+                RA_FRAC_EXCLUDE_TOTAL_UPDATE
+                RA_FRAC_STRIP
+                RA_CONSUMER_PORT
+                RA_CONSUMER_HOST
+                RA_CONSUMER_TOTAL_INIT
+                RA_TIMEOUT
+                RA_CHANGES
+
+
             @return dn_agreement - DN of the created agreement
-                    
+
             @raise InvalidArgumentError - If the suffix is missing
             @raise NosuchEntryError     - if a replica doesn't exist for that suffix
             @raise UNWILLING_TO_PERFORM if the database was previously
                     in read-only state. To create new agreements you
                     need to *restart* the directory server
-            
+
         """
         import string
-        
+
         # Check we have a suffix [ mandatory ]
         if not suffix:
             self.log.warning("create: suffix is missing")
             raise InvalidArgumentError('suffix is mandatory')
-        
+
         if properties:
             binddn      = properties.get(RA_BINDDN)         or defaultProperties[REPLICATION_BIND_DN]
             bindpw      = properties.get(RA_BINDPW)         or defaultProperties[REPLICATION_BIND_PW]
@@ -434,7 +429,7 @@ class Agreement(object):
             description = format
             transport   = defaultProperties[REPLICATION_TRANSPORT]
             timeout     = defaultProperties[REPLICATION_TIMEOUT]
-        
+
         # Compute the normalized suffix to be set in RA entry
         nsuffix = normalizeDN(suffix)
 
@@ -452,7 +447,7 @@ class Agreement(object):
         # This is probably unnecessary because
         # we can just raise ALREADY_EXISTS
         try:
-            
+
             entry = self.conn.getEntry(dn_agreement, ldap.SCOPE_BASE)
             self.log.warn("Agreement already exists: %r" % dn_agreement)
             return dn_agreement
@@ -474,7 +469,7 @@ class Agreement(object):
             RA_PROPNAME_TO_ATTRNAME[RA_METHOD]:         bindmethod,
             RA_PROPNAME_TO_ATTRNAME[RA_DESCRIPTION]:    string.Template(description).substitute({'host': host, 'port': port})
         })
-        
+
         # we make a copy here because we cannot change
         # the passed in properties dict
         propertiescopy = {}
@@ -525,8 +520,6 @@ class Agreement(object):
 
         return dn_agreement
 
-        
-
     def init(self, suffix=None, consumer_host=None, consumer_port=None):
         """Trigger a total update of the consumer replica
             - self is the supplier,
@@ -538,19 +531,19 @@ class Agreement(object):
 
             @raise InvalidArgument: if missing mandatory argurment (suffix/host/port)
         """
-        # 
+        #
         # check the required parameters are set
         #
         if not suffix:
             self.log.fatal("initAgreement: suffix is missing")
             raise InvalidArgumentError('suffix is mandatory argument')
-        
+
         nsuffix = normalizeDN(suffix)
-            
+
         if not consumer_host:
             self.log.fatal("initAgreement: host is missing")
             raise InvalidArgumentError('host is mandatory argument')
-            
+
         if not consumer_port:
             self.log.fatal("initAgreement: port is missing")
             raise InvalidArgumentError('port is mandatory argument')
@@ -569,7 +562,7 @@ class Agreement(object):
         except ldap.NO_SUCH_OBJECT:
             self.log.fatal("initAgreement: No replica agreement to %s:%d for suffix %s" % (consumer_host, consumer_port, nsuffix))
             raise
-            
+
         #
         # trigger the total init
         #
@@ -581,10 +574,10 @@ class Agreement(object):
         """Pause this replication agreement.  This replication agreement
           will send no more changes.  Use the resume() method to "unpause".
           It tries to disable the replica agreement. If it fails (not implemented in all version),
-          it uses the schedule() with interval '2358-2359 0' 
+          it uses the schedule() with interval '2358-2359 0'
             @param agmtdn - agreement dn
             @param interval - (default NEVER) replication schedule to use
-            
+
             @return None
 
             @raise ValueError - if interval is not valid
@@ -594,18 +587,21 @@ class Agreement(object):
             ldap.MOD_REPLACE, 'nsds5ReplicaEnabled', ['off'])]
         try:
             self.conn.modify_s(agmtdn, mod)
-        except ldap.LDAPError, e:
+        except ldap.LDAPError:
             # before 1.2.11, no support for nsds5ReplicaEnabled
             # use schedule hack
             self.schedule(interval)
 
+        # Allow a little time for the change to take effect
+        time.sleep(2)
+
     def resume(self, agmtdn, interval=ALWAYS):
         """Resume a paused replication agreement, paused with the "pause" method.
            It tries to enabled the replica agreement. If it fails (not implemented in all version),
-           it uses the schedule() with interval '0000-2359 0123456' 
+           it uses the schedule() with interval '0000-2359 0123456'
             @param agmtdn  - agreement dn
             @param interval - (default ALWAYS) replication schedule to use
-            
+
             @return None
 
             @raise ValueError - if interval is not valid
@@ -616,17 +612,20 @@ class Agreement(object):
             ldap.MOD_REPLACE, 'nsds5ReplicaEnabled', ['on'])]
         try:
             self.conn.modify_s(agmtdn, mod)
-        except ldap.LDAPError, e:
+        except ldap.LDAPError:
             # before 1.2.11, no support for nsds5ReplicaEnabled
             # use schedule hack
             self.schedule(interval)
 
+        # Allow a little time for the change to take effect
+        time.sleep(2)
+
     def changes(self, agmnt_dn):
         """Return a list of changes sent by this agreement."""
         retval = 0
         try:
             ent = self.conn.getEntry(
-                agmnt_dn, ldap.SCOPE_BASE, "(objectclass=*)", [ RA_PROPNAME_TO_ATTRNAME[RA_CHANGES] ])
+                agmnt_dn, ldap.SCOPE_BASE, "(objectclass=*)", [RA_PROPNAME_TO_ATTRNAME[RA_CHANGES]])
         except:
             raise NoSuchEntryError(
                 "Error reading status from agreement", agmnt_dn)
@@ -642,4 +641,3 @@ class Agreement(object):
                     if ary and len(ary) > 1:
                         retval = retval + int(ary[1].split("/")[0])
         return retval
-        
\ No newline at end of file
-- 
1.9.3

