From 50732c9e8767b60c4bd52b0c48924b5093897dd5 Mon Sep 17 00:00:00 2001 From: Simon Pichugin Date: May 14 2018 07:33:23 +0000 Subject: Issue 49657 - Fix cascading replication scenario in lib389 API Bug description: We should be able to create cascading replication topology with existing lib389 API. Fix existing topology in topologies.py and add the functionality to replica.py API accordingly. Fix description: Add the code to join_hub function. Get agreement name from the last three port numbers. When we call _create_service_group on read-only instance do not try to create it. Generate the replica credentials when adding the service account and store them in ReplicationManager object. Fix a few small issues in replica.py. In topologies.py, divide the instance and replica creation. Refactor topology_m1h1c1 according to the changes. During the finalizer, check if instance exists before the removal. https://pagure.io/389-ds-base/issue/49657 Reviewed by: mreynolds (Thanks!) --- diff --git a/src/lib389/lib389/replica.py b/src/lib389/lib389/replica.py index 0e72310..27bd0f8 100644 --- a/src/lib389/lib389/replica.py +++ b/src/lib389/lib389/replica.py @@ -1296,6 +1296,7 @@ class ReplicationManager(object): else: self._log = logging.getLogger(__name__) self._alloc_rids = [] + self._repl_creds = {} def _ensure_changelog(self, instance): """Internally guarantee a changelog exists for @@ -1314,9 +1315,7 @@ class ReplicationManager(object): """From an instance, determine the agreement name that we would use for it. Internal only. """ - to_replicas = Replicas(to_instance) - to_r = to_replicas.get(self._suffix) - return to_r.get_rid() + return str(to_instance.port)[-3:] def create_first_master(self, instance): """In a topology, this creates the "first" master that has the @@ -1354,12 +1353,31 @@ class ReplicationManager(object): def _create_service_group(self, from_instance): """Internally create the service group that contains replication managers. This may become part of the default objects in the future. Internal only. + + When we join a consumer to hub the function check that the group is in place """ + groups = Groups(from_instance, basedn=self._suffix, rdn=None) - repl_group = groups.ensure_state(properties={ - 'cn': 'replication_managers', - }) - return repl_group + from_replicas = Replicas(from_instance) + try: + from_r = from_replicas.get(self._suffix) + repl_type = from_r.get_attr_val_int('nsDS5ReplicaType') + except ldap.NO_SUCH_OBJECT: + repl_type = None + + if repl_type == 3 or repl_type is None: + repl_group = groups.ensure_state(properties={ + 'cn': 'replication_managers', + }) + return repl_group + else: + try: + repl_group = groups.get('replication_managers') + return repl_group + except ldap.NO_SUCH_OBJECT: + self._log.warning("{} doesn't have cn=replication_managers,{} entry \ + and the instance is not read-write".format(from_instance.serverid, self._suffix)) + raise def _create_service_account(self, from_instance, to_instance): """Create the server replication service account, and @@ -1376,11 +1394,15 @@ class ReplicationManager(object): port = to_instance.sslport services = ServiceAccounts(from_instance, self._suffix) - # We don't have an agreement yet, so don't bother with the - # password yet ... + # Generate the password and save the credentials + # for putting them into agreements in the future + service_name = '{}:{}'.format(to_instance.host, port) + creds = password_generate() repl_service = services.ensure_state(properties={ - 'cn': '%s:%s' % (to_instance.host, port), + 'cn': service_name, + 'userPassword': creds }) + self._repl_creds[service_name] = creds repl_group.ensure_member(repl_service.dn) @@ -1453,14 +1475,14 @@ class ReplicationManager(object): to_replicas = Replicas(to_instance) try: to_r = to_replicas.get(self._suffix) - self._log("WARNING: to_instance is already a replica for this suffix") + self._log.warning("{} is already a replica for this suffix".format(to_instance.serverid)) return except ldap.NO_SUCH_OBJECT: pass # Make sure we replicate this suffix too ... - fr_replicas = Replicas(from_instance) - fr_r = fr_replicas.get(self._suffix) + from_replicas = Replicas(from_instance) + from_r = from_replicas.get(self._suffix) # Ensure we have a cl self._ensure_changelog(to_instance) @@ -1469,7 +1491,7 @@ class ReplicationManager(object): repl_dn = self._create_service_account(from_instance, to_instance) # Find the ruv on from_instance - ruv = fr_r.get_ruv() + ruv = from_r.get_ruv() # Get a free rid rid = ruv.alloc_rid() @@ -1489,15 +1511,15 @@ class ReplicationManager(object): # WARNING: You need to create passwords and agmts BEFORE you tot_init! + # perform the _bootstrap. This creates a temporary repl manager + # to allow the tot_init to occur. + self._bootstrap_replica(from_r, to_r, to_instance) + # Now put in an agreement from to -> from # both ends. self.ensure_agreement(from_instance, to_instance) self.ensure_agreement(to_instance, from_instance, init=True) - # perform the _bootstrap. This creates a temporare repl manager - # to allow the tot_init to occur. - self._bootstrap_replica(fr_r, to_r, to_instance) - # Now fix our replica credentials from -> to to_r.set('nsDS5ReplicaBindDNGroup', repl_dn) @@ -1520,9 +1542,50 @@ class ReplicationManager(object): :param to_instance: An instance to join to the topology. :type to_instance: lib389.DirSrv """ - # Ensure we have a cl + + to_replicas = Replicas(to_instance) + try: + to_r = to_replicas.get(self._suffix) + self._log.warning("{} is already a replica for this suffix".format(to_instance.serverid)) + return + except ldap.NO_SUCH_OBJECT: + pass + + # Make sure we replicate this suffix too ... + from_replicas = Replicas(from_instance) + from_r = from_replicas.get(self._suffix) + + # Ensure we have a changelog self._ensure_changelog(to_instance) - raise Exception + + # Create replica on to_instance, with bootstrap details. + to_r = to_replicas.create(properties={ + 'cn': 'replica', + 'nsDS5ReplicaRoot': self._suffix, + 'nsDS5ReplicaId': '65535', + 'nsDS5Flags': '1', + 'nsDS5ReplicaType': '2', + 'nsds5replicabinddngroupcheckinterval': '0' + }) + + # WARNING: You need to create passwords and agmts BEFORE you tot_init! + repl_dn = self._create_service_account(from_instance, to_instance) + + # perform the _bootstrap. This creates a temporary repl manager + # to allow the tot_init to occur. + self._bootstrap_replica(from_r, to_r, to_instance) + + # Now put in an agreement from to -> from + # both ends. + self.ensure_agreement(from_instance, to_instance) + + # Now fix our replica credentials from -> to + to_r.set('nsDS5ReplicaBindDNGroup', repl_dn) + + # Now finally test it ... + self.test_replication(from_instance, to_instance) + # Done! + self._log.info("SUCCESS: joined consumer from %s to %s" % (from_instance.ldapuri, to_instance.ldapuri)) def join_consumer(self, from_instance, to_instance): """Join a new consumer to this instance. This will complete @@ -1539,14 +1602,14 @@ class ReplicationManager(object): to_replicas = Replicas(to_instance) try: to_r = to_replicas.get(self._suffix) - self._log("WARNING: to_instance is already a replica for this suffix") + self._log.warning("{} is already a replica for this suffix".format(to_instance.serverid)) return except ldap.NO_SUCH_OBJECT: pass # Make sure we replicate this suffix too ... - fr_replicas = Replicas(from_instance) - fr_r = fr_replicas.get(self._suffix) + from_replicas = Replicas(from_instance) + from_r = from_replicas.get(self._suffix) # Create replica on to_instance, with bootstrap details. to_r = to_replicas.create(properties={ @@ -1559,21 +1622,25 @@ class ReplicationManager(object): }) # WARNING: You need to create passwords and agmts BEFORE you tot_init! + # If from_instance replica isn't read-write (hub, probably), we just check it is there repl_group = self._create_service_group(from_instance) + # perform the _bootstrap. This creates a temporary repl manager + # to allow the tot_init to occur. + self._bootstrap_replica(from_r, to_r, to_instance) + # Now put in an agreement from to -> from # both ends. self.ensure_agreement(from_instance, to_instance) - # perform the _bootstrap. This creates a temporare repl manager - # to allow the tot_init to occur. - self._bootstrap_replica(fr_r, to_r, to_instance) - # Now fix our replica credentials from -> to to_r.set('nsDS5ReplicaBindDNGroup', repl_group.dn) # Now finally test it ... - self.test_replication(from_instance, to_instance) + # If from_instance replica isn't read-write (hub, probably), we will test it later + if from_r.get_attr_val_int('nsDS5ReplicaType') == 3: + self.test_replication(from_instance, to_instance) + # Done! self._log.info("SUCCESS: joined consumer from %s to %s" % (from_instance.ldapuri, to_instance.ldapuri)) @@ -1592,33 +1659,14 @@ class ReplicationManager(object): Internal Only. """ - # We write all our changes to "write_instance", but we read data - # from the "from" instance. - - dn = None - creds = None - - fr_replicas = Replicas(from_instance) - fr_r = fr_replicas.get(self._suffix) - from_agmts = fr_r.get_agreements() - # see if any exist already .... - agmts = from_agmts.list() - if len(agmts) > 0: - # okay, re-use the creds - agmt = agmts[0] - dn = agmt.get_attr_val_utf8('nsDS5ReplicaBindDN') - creds = agmt.get_attr_val_utf8('nsDS5ReplicaCredentials') - else: - # Create them ... - # Get the service account. - services = ServiceAccounts(write_instance, self._suffix) - sa = services.get('%s:%s' % (from_instance.host, from_instance.sslport)) - creds = password_generate() - # Gen a password - sa.set('userPassword', creds) - dn = sa.dn - return (dn, creds) + rdn = '{}:{}'.format(from_instance.host, from_instance.sslport) + creds = self._repl_creds[rdn] + + services = ServiceAccounts(write_instance, self._suffix) + sa_dn = services.get(rdn).dn + + return (sa_dn, creds) def ensure_agreement(self, from_instance, to_instance, init=False): """Guarantee that a replication agreement exists 'from_instance' send @@ -1626,7 +1674,7 @@ class ReplicationManager(object): consumer. Both instances must have been added to the topology with - create first master, join_master or join_consumer. + create first master, join_master, join_consumer or join_hub. :param from_instance: An instance already in the topology. :type from_instance: lib389.DirSrv @@ -1643,17 +1691,17 @@ class ReplicationManager(object): # init = False (default) means creds *might* exist, and we create them # on the "from" master. - fr_replicas = Replicas(from_instance) - fr_r = fr_replicas.get(self._suffix) + from_replicas = Replicas(from_instance) + from_r = from_replicas.get(self._suffix) - from_agmts = fr_r.get_agreements() + from_agmts = from_r.get_agreements() agmt_name = self._inst_to_agreement_name(to_instance) try: agmt = from_agmts.get(agmt_name) self._log.info("SUCCESS: Agreement from %s to %s already exists" % (from_instance.ldapuri, to_instance.ldapuri)) - return + return agmt except ldap.NO_SUCH_OBJECT: # Okay, it doesn't exist, lets go ahead! pass @@ -1720,10 +1768,10 @@ class ReplicationManager(object): # No agreement, that's good! pass - fr_replicas = Replicas(instance) - fr_r = fr_replicas.get(self._suffix) + from_replicas = Replicas(instance) + from_r = from_replicas.get(self._suffix) # This should delete the agreements .... - fr_r.delete() + from_r.delete() def disable_to_master(self, to_instance, from_instances=[]): """For all masters "from" disable all agreements "to" instance. @@ -1769,13 +1817,13 @@ class ReplicationManager(object): :type to_instance: lib389.DirSrv """ - fr_replicas = Replicas(from_instance) - fr_r = fr_replicas.get(self._suffix) + from_replicas = Replicas(from_instance) + from_r = from_replicas.get(self._suffix) to_replicas = Replicas(to_instance) to_r = to_replicas.get(self._suffix) - from_ruv = fr_r.get_ruv() + from_ruv = from_r.get_ruv() for i in range(0, timeout): to_ruv = to_r.get_ruv() @@ -1815,8 +1863,6 @@ class ReplicationManager(object): time.sleep(1) raise Exception("Replication did not sync in time!") - self.wait_for_replication(from_instance, to_instance) - def test_replication(self, from_instance, to_instance, timeout=20): """Wait for a replication event to occur from instance to instance. This diff --git a/src/lib389/lib389/topologies.py b/src/lib389/lib389/topologies.py index 690c7f6..a41df3f 100644 --- a/src/lib389/lib389/topologies.py +++ b/src/lib389/lib389/topologies.py @@ -8,7 +8,6 @@ # import os import logging -import time # For hostname detection for GSSAPI tests import socket @@ -16,15 +15,11 @@ import socket import pytest from lib389 import DirSrv -from lib389.nss_ssl import NssSsl from lib389.utils import generate_ds_params from lib389.mit_krb5 import MitKrb5 from lib389.saslmap import SaslMappings from lib389.replica import ReplicationManager, Replicas - -from lib389._constants import (SER_HOST, SER_PORT, SER_SERVERID_PROP, SER_CREATION_SUFFIX, - SER_SECURE_PORT, ReplicaRole, DEFAULT_SUFFIX, REPLICA_ID, - SER_LDAP_URL) +from lib389._constants import * DEBUGGING = os.getenv('DEBUGGING', default=False) if DEBUGGING: @@ -34,30 +29,23 @@ else: log = logging.getLogger(__name__) -def create_topology(topo_dict, suffix=DEFAULT_SUFFIX): - """Create a requested topology. Cascading replication scenario isn't supported +def _create_instances(topo_dict, suffix): + """Create requested instances without replication or any other modifications :param topo_dict: a dictionary {ReplicaRole.STANDALONE: num, ReplicaRole.MASTER: num, - ReplicaRole.CONSUMER: num} + ReplicaRole.HUB: num, ReplicaRole.CONSUMER: num} :type topo_dict: dict - :param suffix: a suffix for the replication + :param suffix: a suffix :type suffix: str :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 = {} cs = {} + hs = {} ins = {} - replica_dict = {} # Create instances for role in topo_dict.keys(): @@ -103,12 +91,42 @@ def create_topology(topo_dict, suffix=DEFAULT_SUFFIX): if role == ReplicaRole.CONSUMER: cs[instance.serverid] = instance instances.update(cs) + if role == ReplicaRole.HUB: + hs[instance.serverid] = instance + instances.update(hs) log.info("Instance with parameters {} was created.".format(args_instance)) + if "standalone1" in instances and len(instances) == 1: + return TopologyMain(standalones=instances["standalone1"]) + else: + return TopologyMain(standalones=ins, masters=ms, consumers=cs, hubs=hs) + + +def create_topology(topo_dict, suffix=DEFAULT_SUFFIX): + """Create a requested topology. Cascading replication scenario isn't supported + + :param topo_dict: a dictionary {ReplicaRole.STANDALONE: num, ReplicaRole.MASTER: num, + ReplicaRole.CONSUMER: num} + :type topo_dict: dict + :param suffix: a suffix for the replication + :type suffix: str + + :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.") + + topo = _create_instances(topo_dict, suffix) + # Start with a single master, and create it "first". first_master = None try: - first_master = list(ms.values())[0] + first_master = list(topo.ms.values())[0] log.info("Creating replication topology.") # Now get the first master ready. repl = ReplicationManager(DEFAULT_SUFFIX) @@ -119,7 +137,7 @@ def create_topology(topo_dict, suffix=DEFAULT_SUFFIX): # Now init the other masters from this. # This will reinit m, and put a bi-directional agreement # in place. - for m in ms.values(): + for m in topo.ms.values(): # Skip firstmaster. if m is first_master: continue @@ -127,35 +145,35 @@ def create_topology(topo_dict, suffix=DEFAULT_SUFFIX): repl.join_master(first_master, m) # Mesh the master agreements. - for mo in ms.values(): - for mi in ms.values(): + for mo in topo.ms.values(): + for mi in topo.ms.values(): if mo is mi: continue log.info("Ensuring master %s to %s ..." % (mo.serverid, mi.serverid)) repl.ensure_agreement(mo, mi) # Add master -> consumer agreements. - for c in cs.values(): - log.info("Joining consumer %s from %s ..." % (mo.serverid, mi.serverid)) + for c in topo.cs.values(): + log.info("Joining consumer %s from %s ..." % (c.serverid, first_master.serverid)) repl.join_consumer(first_master, c) - for m in ms.values(): - for c in cs.values(): + for m in topo.ms.values(): + for c in topo.cs.values(): log.info("Ensuring consumer %s from %s ..." % (c.serverid, m.serverid)) repl.ensure_agreement(m, c) # Clear out the tmp dir - for instance in instances.values(): + for instance in topo: instance.clearTmpDir(__file__) - if "standalone1" in instances and len(instances) == 1: - return TopologyMain(standalones=instances["standalone1"]) - else: - return TopologyMain(standalones=ins, masters=ms, consumers=cs) + return topo class TopologyMain(object): def __init__(self, standalones=None, masters=None, consumers=None, hubs=None): + self.ms = {} + self.cs = {} + self.hs = {} self.all_insts = {} if standalones: @@ -214,7 +232,8 @@ def topology_st(request): if DEBUGGING: topology.standalone.stop() else: - topology.standalone.delete() + if topology.standalone.exists(): + topology.standalone.delete() request.addfinalizer(fin) return topology @@ -273,11 +292,14 @@ def topology_st_gssapi(request): topology.standalone.restart() + topology.standalone.clearTmpDir(__file__) + def fin(): if DEBUGGING: topology.standalone.stop() else: - topology.standalone.delete() + if topology.standalone.exists(): + topology.standalone.delete() krb.destroy_realm() request.addfinalizer(fin) @@ -295,7 +317,7 @@ def topology_i2(request): if DEBUGGING: [inst.stop() for inst in topology] else: - [inst.delete() for inst in topology] + [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) return topology @@ -311,7 +333,7 @@ def topology_i3(request): if DEBUGGING: [inst.stop() for inst in topology] else: - [inst.delete() for inst in topology] + [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) return topology @@ -326,7 +348,7 @@ def topology_m1(request): if DEBUGGING: [inst.stop() for inst in topology] else: - [inst.delete() for inst in topology] + [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) return topology @@ -342,7 +364,7 @@ def topology_m1c1(request): if DEBUGGING: [inst.stop() for inst in topology] else: - [inst.delete() for inst in topology] + [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) return topology @@ -358,7 +380,7 @@ def topology_m2(request): if DEBUGGING: [inst.stop() for inst in topology] else: - [inst.delete() for inst in topology] + [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) return topology @@ -374,7 +396,7 @@ def topology_m3(request): if DEBUGGING: [inst.stop() for inst in topology] else: - [inst.delete() for inst in topology] + [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) return topology @@ -390,7 +412,7 @@ def topology_m4(request): if DEBUGGING: [inst.stop() for inst in topology] else: - [inst.delete() for inst in topology] + [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) return topology @@ -407,7 +429,7 @@ def topology_m2c2(request): if DEBUGGING: [inst.stop() for inst in topology] else: - [inst.delete() for inst in topology] + [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) return topology @@ -417,76 +439,32 @@ def topology_m2c2(request): 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 = {} + topo_roles = {ReplicaRole.MASTER: 1, ReplicaRole.HUB: 1, ReplicaRole.CONSUMER: 1} + topology = _create_instances(topo_roles, DEFAULT_SUFFIX) + master = topology.ms["master1"] + hub = topology.hs["hub1"] + consumer = topology.cs["consumer1"] - # 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 = {} - args_instance[SER_PORT] = instance_data[SER_PORT] - args_instance[SER_SECURE_PORT] = instance_data[SER_SECURE_PORT] - args_instance[SER_SERVERID_PROP] = instance_data[SER_SERVERID_PROP] - args_instance[SER_CREATION_SUFFIX] = DEFAULT_SUFFIX - instance.allocate(args_instance) - instance_exists = instance.exists() - if instance_exists: - instance.delete() - instance.create() - instance.open() - log.info("Instance with parameters {} was created.".format(args_instance)) - - # 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) - assert replicas.test(DEFAULT_SUFFIX, consumer) + # Start with the master, and create it "first". + log.info("Creating replication topology.") + # Now get the first master ready. + repl = ReplicationManager(DEFAULT_SUFFIX) + repl.create_first_master(master) + # Finish the topology creation + repl.join_hub(master, hub) + repl.join_consumer(hub, consumer) + + repl.test_replication(master, consumer) # Clear out the tmp dir - master.clearTmpDir(__file__) + for instance in topology: + instance.clearTmpDir(__file__) def fin(): if DEBUGGING: - [inst.stop() for inst in instances] + [inst.stop() for inst in topology] else: - [inst.delete() for inst in instances] + [inst.delete() for inst in topology if inst.exists()] request.addfinalizer(fin) - return TopologyMain(masters={"master1": master}, hubs={"hub1": hub}, consumers={"consumer1": consumer}) - + return topology