From 3baa7995a816707f95cc0dd7b29d8bc6f126711b Mon Sep 17 00:00:00 2001 From: Simon Pichugin Date: Fri, 11 Aug 2017 23:43:06 +0200 Subject: [PATCH] Issue 91 - Fix replication m1h1c1 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: ? 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 - - -@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 + + +@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}) + -- 2.13.3