This is an automated email from the git hooks/post-receive script.
spichugi pushed a commit to branch master
in repository lib389.
commit 3ee4f1cd64eeed7f057111dc3ece52ba3f77c36d
Author: Simon Pichugin <spichugi(a)redhat.com>
Date: Fri Aug 11 23:43:06 2017 +0200
Issue 91 - Fix replication topology
Bug Description: When you use m1h1c1 topologies the replica type is
incorrectly set for consumers(maybe the others too), and it adds repl
agreements to consumers. Also there is circular repl initialization
going on which causes everybody to have different db generation IDs
which then breaks replication.
Fix Description: Refactor create_topology(), make it more readable
and depricate cascading replication scenarion. Let's create it when we
need it.
Fix an issue with creating agreements on consumers.
Fix an issue with agreements initialization.
Remade topology_m1h1c1 so it wouldn't use create_topology() func.
Fix small replica.py issues.
Reviewed by: wibrown, mreynolds (Thanks!)
https://pagure.io/lib389/issue/91
---
lib389/replica.py | 4 +-
lib389/topologies.py | 150 ++++++++++++++++++++++++++++++++++++---------------
2 files changed, 108 insertions(+), 46 deletions(-)
diff --git a/lib389/replica.py b/lib389/replica.py
index 9251ecd..a028396 100644
--- a/lib389/replica.py
+++ b/lib389/replica.py
@@ -1055,7 +1055,7 @@ class Replica(DSLdapObject):
entry = self._instance.getEntry(
agmtdn, ldap.SCOPE_BASE, "(objectclass=*)", attrlist)
except NoSuchEntryError:
- self.log.exception("Error reading status from agreement
{}".format(agmtdn))
+ self._log.exception("Error reading status from agreement
{}".format(agmtdn))
hasError = 1
else:
refresh = entry.nsds5BeginReplicaRefresh
@@ -1121,7 +1121,7 @@ class Replica(DSLdapObject):
THIS SHOULD BE IN THE NEW AGREEMENT CLASS
"""
- self.log.info("Starting async replication %s" % agmtdn)
+ self._log.info("Starting async replication %s" % agmtdn)
mod = [(ldap.MOD_ADD, 'nsds5BeginReplicaRefresh', 'start')]
self._instance.modify_s(agmtdn, mod)
diff --git a/lib389/topologies.py b/lib389/topologies.py
index 225e138..1ffc09e 100644
--- a/lib389/topologies.py
+++ b/lib389/topologies.py
@@ -27,22 +27,25 @@ log = logging.getLogger(__name__)
def create_topology(topo_dict):
- """Create a requested topology
+ """Create a requested topology. Cascading replication scenario
isn't supported
- @param topo_dict - dictionary {REPLICAROLE: number_of_insts}
- @return - dictionary {serverid: topology_instance}
+ @param topo_dict - dictionary {ReplicaRole.STANDALONE: num, ReplicaRole.MASTER: num,
+ ReplicaRole.CONSUMER: num}
+ @return - TopologyMain object
"""
if not topo_dict:
ValueError("You need to specify the dict. For instance:
{ReplicaRole.STANDALONE: 1}")
+ if ReplicaRole.HUB in topo_dict.keys():
+ NotImplementedError("Cascading replication scenario isn't
supported."
+ "Please, use existing topology or create your
own.")
+
instances = {}
ms = {}
- hs = {}
cs = {}
ins = {}
replica_dict = {}
- agmts = {}
# Create instances
for role in topo_dict.keys():
@@ -73,48 +76,48 @@ def create_topology(topo_dict):
if role == ReplicaRole.MASTER:
ms[instance.serverid] = instance
instances.update(ms)
- if role == ReplicaRole.HUB:
- hs[instance.serverid] = instance
- instances.update(hs)
if role == ReplicaRole.CONSUMER:
cs[instance.serverid] = instance
instances.update(cs)
log.info("Instance with parameters {} was
created.".format(args_copied))
# Set up replication
- if role in (ReplicaRole.MASTER, ReplicaRole.HUB, ReplicaRole.CONSUMER):
+ if role in (ReplicaRole.MASTER, ReplicaRole.CONSUMER):
replicas = Replicas(instance)
replica = replicas.enable(DEFAULT_SUFFIX, role,
instance_data[REPLICA_ID])
replica_dict[replica] = instance
- # Create agreements
for role_from in topo_dict.keys():
+ # Do not create agreements on consumer
+ if role_from == ReplicaRole.CONSUMER:
+ continue
+
+ # Create agreements: master -> masters, consumers
for inst_num_from in range(1, topo_dict[role]+1):
- roles_to = [ReplicaRole.HUB, ReplicaRole.CONSUMER]
- if role == ReplicaRole.MASTER:
- roles_to.append(ReplicaRole.MASTER)
+ roles_to = [ReplicaRole.MASTER, ReplicaRole.CONSUMER]
for role_to in [role for role in topo_dict if role in roles_to]:
for inst_num_to in range(1, topo_dict[role]+1):
- # Exclude our instance
+ # Exclude the instance we created it from
if role_from != role_to or inst_num_from != inst_num_to:
inst_from_id = "{}{}".format(role_from.name.lower(),
inst_num_from)
inst_to_id = "{}{}".format(role_to.name.lower(),
inst_num_to)
inst_from = instances[inst_from_id]
inst_to = instances[inst_to_id]
- agmt = inst_from.agreement.create(suffix=DEFAULT_SUFFIX,
- host=inst_to.host,
- port=inst_to.port)
- agmts[agmt] = (inst_from, inst_to)
+ inst_from.agreement.create(suffix=DEFAULT_SUFFIX,
+ host=inst_to.host,
+ port=inst_to.port)
# Allow the replicas to get situated with the new agreements
- if agmts:
+ if replica_dict:
time.sleep(5)
- # Initialize all the agreements
- for agmt, insts in agmts.items():
- insts[0].agreement.init(DEFAULT_SUFFIX, insts[1].host, insts[1].port)
- insts[0].waitForReplInit(agmt)
+ # Initialize all agreements of one master (consumers)
+ for replica_from, inst_from in replica_dict.items():
+ if replica_from.get_role() == ReplicaRole.MASTER:
+ agmts = inst_from.agreement.list(DEFAULT_SUFFIX)
+ map(lambda agmt: replica_from.start_and_wait(agmt.dn), agmts)
+ break
# Clear out the tmp dir
for instance in instances.values():
@@ -123,7 +126,7 @@ def create_topology(topo_dict):
if "standalone1" in instances and len(instances) == 1:
return TopologyMain(standalones=instances["standalone1"])
else:
- return TopologyMain(standalones=ins, masters=ms, hubs=hs, consumers=cs)
+ return TopologyMain(standalones=ins, masters=ms, consumers=cs)
class TopologyMain(object):
@@ -228,26 +231,6 @@ def topology_m1c1(request):
@pytest.fixture(scope="module")
-def topology_m1h1c1(request):
- """Create Replication Deployment with one master, one consumer and one
hub"""
-
- topology = create_topology({ReplicaRole.MASTER: 1,
- ReplicaRole.HUB: 1,
- ReplicaRole.CONSUMER: 1})
- replicas = Replicas(topology.ms["master1"])
- replicas.test(DEFAULT_SUFFIX, topology.cs["consumer1"])
-
- def fin():
- if DEBUGGING:
- map(lambda inst: inst.stop(), topology.all_insts.values())
- else:
- map(lambda inst: inst.delete(), topology.all_insts.values())
- request.addfinalizer(fin)
-
- return topology
-
-
-(a)pytest.fixture(scope="module")
def topology_m2(request):
"""Create Replication Deployment with two masters"""
@@ -318,3 +301,82 @@ def topology_m2c2(request):
request.addfinalizer(fin)
return topology
+
+
+(a)pytest.fixture(scope="module")
+def topology_m1h1c1(request):
+ """Create Replication Deployment with one master, one consumer and one
hub"""
+
+ roles = (ReplicaRole.MASTER, ReplicaRole.HUB, ReplicaRole.CONSUMER)
+ instances = []
+ replica_dict = {}
+
+ # Create instances
+ for role in roles:
+ instance_data = generate_ds_params(1, role)
+ if DEBUGGING:
+ instance = DirSrv(verbose=True)
+ else:
+ instance = DirSrv(verbose=False)
+ args_instance[SER_HOST] = instance_data[SER_HOST]
+ args_instance[SER_PORT] = instance_data[SER_PORT]
+ args_instance[SER_SERVERID_PROP] = instance_data[SER_SERVERID_PROP]
+ args_instance[SER_CREATION_SUFFIX] = DEFAULT_SUFFIX
+ args_copied = args_instance.copy()
+ instance.allocate(args_copied)
+ instance_exists = instance.exists()
+ if instance_exists:
+ instance.delete()
+ instance.create()
+ instance.open()
+ log.info("Instance with parameters {} was
created.".format(args_copied))
+
+ # Set up replication
+ replicas = Replicas(instance)
+ replica = replicas.enable(DEFAULT_SUFFIX, role, instance_data[REPLICA_ID])
+
+ if role == ReplicaRole.MASTER:
+ master = instance
+ replica_master = replica
+ instances.append(master)
+ if role == ReplicaRole.HUB:
+ hub = instance
+ replica_hub = replica
+ instances.append(hub)
+ if role == ReplicaRole.CONSUMER:
+ consumer = instance
+ instances.append(consumer)
+
+ # Create all the agreements
+ # Creating agreement from master to hub
+ master.agreement.create(suffix=DEFAULT_SUFFIX, host=hub.host, port=hub.port)
+
+ # Creating agreement from hub to consumer
+ hub.agreement.create(suffix=DEFAULT_SUFFIX, host=consumer.host, port=consumer.port)
+
+ # Allow the replicas to get situated with the new agreements...
+ time.sleep(5)
+
+ # Initialize all the agreements
+ agmt = master.agreement.list(DEFAULT_SUFFIX)[0].dn
+ replica_master.start_and_wait(agmt)
+
+ agmt = hub.agreement.list(DEFAULT_SUFFIX)[0].dn
+ replica_hub.start_and_wait(agmt)
+
+ # Check replication is working...
+ replicas = Replicas(master)
+ replicas.test(DEFAULT_SUFFIX, consumer)
+
+ # Clear out the tmp dir
+ master.clearTmpDir(__file__)
+
+ def fin():
+ if DEBUGGING:
+ map(lambda inst: inst.stop(), instances)
+ else:
+ map(lambda inst: inst.delete(), instances)
+ request.addfinalizer(fin)
+
+ return TopologyMain(masters={"master1": master}, hubs={"hub1":
hub}, consumers={"consumer1": consumer})
+
--
To stop receiving notification emails like this one, please contact
the administrator of this repository.