def test_base_dir_win32(self):
        coord = coordination.get_coordinator(
            'file:///C:/path/', self._FAKE_MEMBER_ID)
        self.assertEqual('C:\\path\\', coord._dir)

        coord = coordination.get_coordinator(
            'file:////share_addr/share_path/', self._FAKE_MEMBER_ID)
        self.assertEqual('\\\\share_addr\\share_path\\', coord._dir)

        # Administrative shares should be handled properly.
        coord = coordination.get_coordinator(
            'file:////c$/path/', self._FAKE_MEMBER_ID)
        self.assertEqual('\\\\c$\\path\\', coord._dir)

def coordinator_setup(start_heart=True):
    """
    Sets up the client for the coordination service.

    URL examples for connection:
        zake://
        file:///tmp
        redis://username:[email protected]:port
        mysql://username:[email protected]:port/dbname
    """
    url = cfg.CONF.coordination.url
    lock_timeout = cfg.CONF.coordination.lock_timeout
    member_id = get_member_id()

    if url:
        coordinator = coordination.get_coordinator(url, member_id, lock_timeout=lock_timeout)
    else:
        # Use a no-op backend
        # Note: We don't use tooz to obtain a reference since for this to work we would need to
        # register a plugin inside setup.py entry_point and use python setup.py develop for tests
        # to work
        coordinator = NoOpDriver(member_id)

    coordinator.start(start_heart=start_heart)
    return coordinator

def tooz_make_change(driver, url, session, vol_id, initial, destination,
                     attach_status):
    global coordinator
    global lock
    global acquired

    # If coordinator is not the one we want we cannot reuse it
    if not isinstance(coordinator, driver):
        if coordinator:
            coordinator.stop()

        # Create new coordinator and lock
        coordinator = coordination.get_coordinator(url, str(session))
        coordinator.start()
        lock = coordinator.get_lock(vol_id)

    # When going from available to any other state we acquire the lock
    if initial == 'available':
        # If this is a retry we've already acquired the lock
        if not acquired:
            while not lock.acquire():
                coordinator.heartbeat()
                time.sleep(0.01)
            acquired = True

    n = 0
    while n == 0:
        n = safe_update(session, vol_id,
                        {'status': destination,
                         'attach_status': attach_status},
                        {'status': initial})
    coordinator.heartbeat()
    if destination == 'available':
        lock.release()
        acquired = False

    def __init__(self, worker_id):
        self._worker_id = worker_id
        super(Orchestrator, self).__init__(self._worker_id)

        self.fetcher = driver.DriverManager(
            FETCHERS_NAMESPACE,
            CONF.fetcher.backend,
            invoke_on_load=True,
        ).driver

        transformers = transformer.get_transformers()
        self.collector = collector.get_collector(transformers)
        self.storage = storage.get_storage()
        self._state = state.StateManager()

        # RPC
        self.server = None
        self._rating_endpoint = RatingEndpoint(self)
        self._init_messaging()

        # DLM
        self.coord = coordination.get_coordinator(
            CONF.orchestrator.coordination_url,
            uuidutils.generate_uuid().encode('ascii'))
        self.coord.start(start_heart=True)

def coordinator_setup():
    """
    Sets up the client for the coordination service.

    URL examples for connection:
        zake://
        file:///tmp
        redis://username:[email protected]:port
        mysql://username:[email protected]:port/dbname
    """
    url = cfg.CONF.coordination.url
    lock_timeout = cfg.CONF.coordination.lock_timeout
    proc_info = system_info.get_process_info()
    member_id = six.b('%s_%d' % (proc_info['hostname'], proc_info['pid']))

    if url:
        coordinator = coordination.get_coordinator(url, member_id, lock_timeout=lock_timeout)
    else:
        # Use a no-op backend
        # Note: We don't use tooz to obtain a reference since for this to work we would need to
        # register a plugin inside setup.py entry_point and use python setup.py develop for tests
        # to work
        coordinator = NoOpDriver(member_id)

    coordinator.start()
    return coordinator

 def test_client_failure_heartbeat(self, mock_client_cls):
     mock_client = mock.MagicMock()
     mock_client_cls.return_value = mock_client
     member_id = str(uuid.uuid4()).encode('ascii')
     coord = coordination.get_coordinator(self.FAKE_URL, member_id)
     coord.start()
     mock_client.set.side_effect = socket.timeout('timed-out')
     self.assertRaises(coordination.ToozConnectionError, coord.heartbeat)

    def _get_lock_distributed(name):
        if not LockManager._coordinator:
            LOG.debug("Initialized coordinator with connect string %s", LockManager._connect_string)
            LockManager._coordinator = coordination.get_coordinator(
                LockManager._connect_string, "vmware-neutron-plugin"
            )

        LOG.debug("Retrieved lock for ", name)
        return LockManager._coordinator.get_lock(name)

 def test_client_run_watchers_mixin(self, mock_client_cls,
                                    mock_run_watchers):
     mock_client = mock.MagicMock()
     mock_client_cls.return_value = mock_client
     member_id = tests.get_random_uuid()
     coord = coordination.get_coordinator(self.FAKE_URL, member_id)
     coord.start()
     coord.run_watchers()
     self.assertTrue(mock_run_watchers.called)

 def test_client_failure_leave(self, mock_client_cls):
     mock_client = mock.MagicMock()
     mock_client_cls.return_value = mock_client
     member_id = tests.get_random_uuid()
     coord = coordination.get_coordinator(self.FAKE_URL, member_id)
     coord.start()
     mock_client.gets.side_effect = socket.timeout('timed-out')
     fut = coord.leave_group(tests.get_random_uuid())
     self.assertRaises(coordination.ToozConnectionError, fut.get)

 def test_client_run_watchers_mixin(self, mock_client_cls,
                                    mock_run_watchers):
     mock_client = mock.MagicMock()
     mock_client_cls.return_value = mock_client
     member_id = str(uuid.uuid4()).encode('ascii')
     coord = coordination.get_coordinator(self.FAKE_URL, member_id)
     coord.start()
     coord.run_watchers()
     self.assertTrue(mock_run_watchers.called)

    def start(self):
        if self.started:
            return

        # NOTE(bluex): Tooz expects member_id as a byte string.
        member_id = (self.prefix + self.agent_id).encode('ascii')
        self.coordinator = coordination.get_coordinator(
            cfg.CONF.coordination.backend_url, member_id)
        self.coordinator.start(start_heart=True)
        self.started = True

 def __init__(self, conf):
     super(CarbonaraBasedStorage, self).__init__(conf)
     try:
         self.coord = coordination.get_coordinator(
             conf.coordination_url,
             str(uuid.uuid4()).encode('ascii'))
         self.coord.start(start_heart=True)
     except Exception as e:
         raise storage.StorageError("Unable to start coordinator: %s" % e)
     self.aggregation_workers_number = conf.aggregation_workers_number

    def __init__(self, worker_id, conf, namespaces=None):
        namespaces = namespaces or ['compute', 'central']
        group_prefix = conf.polling.partitioning_group_prefix

        super(AgentManager, self).__init__(worker_id)

        self.conf = conf

        if type(namespaces) is not list:
            namespaces = [namespaces]

        # we'll have default ['compute', 'central'] here if no namespaces will
        # be passed
        extensions = (self._extensions('poll', namespace, self.conf).extensions
                      for namespace in namespaces)
        # get the extensions from pollster builder
        extensions_fb = (self._extensions_from_builder('poll', namespace)
                         for namespace in namespaces)

        self.extensions = list(itertools.chain(*list(extensions))) + list(
            itertools.chain(*list(extensions_fb)))

        if not self.extensions:
            LOG.warning('No valid pollsters can be loaded from %s '
                        'namespaces', namespaces)

        discoveries = (self._extensions('discover', namespace,
                                        self.conf).extensions
                       for namespace in namespaces)
        self.discoveries = list(itertools.chain(*list(discoveries)))
        self.polling_periodics = None

        self.hashrings = None
        self.partition_coordinator = None
        if self.conf.coordination.backend_url:
            # XXX uuid4().bytes ought to work, but it requires ascii for now
            coordination_id = str(uuid.uuid4()).encode('ascii')
            self.partition_coordinator = coordination.get_coordinator(
                self.conf.coordination.backend_url, coordination_id)

        # Compose coordination group prefix.
        # We'll use namespaces as the basement for this partitioning.
        namespace_prefix = '-'.join(sorted(namespaces))
        self.group_prefix = ('%s-%s' % (namespace_prefix, group_prefix)
                             if group_prefix else namespace_prefix)

        self.notifier = oslo_messaging.Notifier(
            messaging.get_transport(self.conf),
            driver=self.conf.publisher_notifier.telemetry_driver,
            publisher_id="ceilometer.polling")

        self._keystone = None
        self._keystone_last_exception = None

 def __init__(self, conf):
     super(CarbonaraBasedStorage, self).__init__(conf)
     self.coord = coordination.get_coordinator(
         conf.coordination_url,
         str(uuid.uuid4()).encode('ascii'))
     self.coord.start()
     if conf.aggregation_workers_number is None:
         try:
             self.aggregation_workers_number = multiprocessing.cpu_count()
         except NotImplementedError:
             self.aggregation_workers_number = 2
     else:
         self.aggregation_workers_number = conf.aggregation_workers_number

    def __init__(self, backend_url):
        self.coordinator = None
        self.member_id = uuidutils.generate_uuid()

        if backend_url:
            try:
                self.coordinator = coordination.get_coordinator(
                    backend_url, self.member_id)
                self.coordinator.start()
                LOG.info('Coordination backend loaded successfully.')
            except coordination.ToozError:
                LOG.error('Error connecting to coordination backend.')
                raise

    def _create_coordinator(self, url):

        def _safe_stop(coord):
            try:
                coord.stop()
            except coordination.ToozError as e:
                message = encodeutils.exception_to_unicode(e)
                if (message != 'Can not stop a driver which has not'
                               ' been started'):
                    raise

        coord = coordination.get_coordinator(url,
                                             tests.get_random_uuid())
        self.addCleanup(_safe_stop, coord)
        return coord

    def _create_coordinator(self, url):

        def _safe_stop(coord):
            try:
                coord.stop()
            except coordination.ToozError as e:
                message = utils.exception_message(e)
                if (message != 'Can not stop a driver which has not'
                               ' been started'):
                    raise

        coord = coordination.get_coordinator(url,
                                             str(uuid.uuid4()).encode('ascii'))
        self.addCleanup(_safe_stop, coord)
        return coord

    def _get_lock_distributed(name):
        if LockManager._coordinator_pid != os.getpid():
            # We should use a per-process coordinator. If PID is different
            # start a new coordinator.
            # While the API workers are spawned, we have to re-initialize
            # a coordinator, so we validate that the PID is still the same.
            LockManager._coordinator_pid = os.getpid()
            LOG.debug('Initialized coordinator with connect string %s',
                      LockManager._connect_string)
            LockManager._coordinator = coordination.get_coordinator(
                LockManager._connect_string, 'vmware-neutron-plugin')
            LockManager._coordinator.start()

        LOG.debug('Retrieved lock for %s', name)
        return LockManager._coordinator.get_lock(name)

def coordinator_setup():
    """
    Sets up the client for the coordination service.

    URL examples for connection:
        zake://
        file:///tmp
        redis://username:[email protected]:port
        mysql://username:[email protected]:port/dbname
    """
    url = cfg.CONF.coordination.url
    lock_timeout = cfg.CONF.coordination.lock_timeout
    proc_info = system_info.get_process_info()
    member_id = '%s_%d' % (proc_info['hostname'], proc_info['pid'])
    coordinator = tooz_coord.get_coordinator(url, member_id, lock_timeout=lock_timeout)
    coordinator.start()
    return coordinator

    def _create_coordinator(self):

        def _safe_stop(coord):
            try:
                coord.stop()
            except coordination.ToozError as e:
                # TODO(harlowja): make this better, so that we don't have to
                # do string checking...
                message = utils.exception_message(e)
                if (message != 'Can not stop a driver which has not'
                               ' been started'):
                    raise

        coord = coordination.get_coordinator(self.FAKE_URL,
                                             str(uuid.uuid4()).encode('ascii'))
        self.addCleanup(_safe_stop, coord)
        return coord