In case of standby cluster this will tel us from which remote
            master to stream. Config can be both patroni config or
            cluster.config.data
332633
def get_remote_member(self, member=None):
        
        cluster_params = self.get_standby_cluster_config()

        if cluster_params:
            name = member.name if member else 'remote_master:{}'.format(uuid.uuid1())

            data = {k: v for k, v in cluster_params.items() if k in RemoteMember.allowed_keys()}
            data['no_replication_slot'] = 'primary_slot_name' not in cluster_params
            conn_kwargs = member.conn_kwargs() if member else \
                {k: cluster_params[k] for k in ('host', 'port') if k in cluster_params}
            if conn_kwargs:
                data['conn_kwargs'] = conn_kwargs

            return RemoteMember(name, data)
498781
public static ClusterInfo discoverClusterInfo(Settings settings, Log log) {
        ClusterName remoteClusterName = null;
        EsMajorVersion remoteVersion = null;
        String clusterName = settings.getProperty(InternalConfigurationOptions.INTERNAL_ES_CLUSTER_NAME);
        String clusterUUID = settings.getProperty(InternalConfigurationOptions.INTERNAL_ES_CLUSTER_UUID);
        String version = settings.getProperty(InternalConfigurationOptions.INTERNAL_ES_VERSION);
        if (StringUtils.hasText(clusterName) && StringUtils.hasText(version)) { // UUID is optional for now
            if (log.isDebugEnabled()) {
                log.debug(String.format("Elasticsearch cluster [NAME:%s][UUID:%s][VERSION:%s] already present in configuration; skipping discovery",
                        clusterName, clusterUUID, version));
            }
            remoteClusterName = new ClusterName(clusterName, clusterUUID);
            remoteVersion = EsMajorVersion.parse(version);
            return new ClusterInfo(remoteClusterName, remoteVersion);
        }

        RestClient bootstrap = new RestClient(settings);
        // first get ES main action info
        try {
            ClusterInfo mainInfo = bootstrap.mainInfo();
            if (log.isDebugEnabled()) {
                log.debug(String.format("Discovered Elasticsearch cluster [%s/%s], version [%s]",
                        mainInfo.getClusterName().getName(),
                        mainInfo.getClusterName().getUUID(),
                        mainInfo.getMajorVersion()));
            }
            settings.setInternalClusterInfo(mainInfo);
            return mainInfo;
        } catch (EsHadoopException ex) {
            throw new EsHadoopIllegalArgumentException(String.format("Cannot detect ES version - "
                    + "typically this happens if the network/Elasticsearch cluster is not accessible or when targeting "
                    + "a WAN/Cloud instance without the proper setting '%s'", ConfigurationOptions.ES_NODES_WAN_ONLY), ex);
        } finally {
            bootstrap.close();
        }
    }
536762
function() {
      var t = this,
          map = {},
          router = Monitor.getRouter(),
          hostName = Monitor.getRouter().getHostName(),
          appName = Monitor.Config.Monitor.appName,
          appInstance = process.env.NODE_APP_INSTANCE;

      // Add this process to the map
      map[hostName] = {};
      map[hostName][appName] = {
        instances: [appInstance],
        probeClasses: _.keys(Probe.classes)
      };

      // Process all known connections
      var connections = router.findConnections();
      connections.forEach(function(connection) {
        hostName = connection.get('hostName') || connection.get('remoteHostName');
        appName = connection.get('remoteAppName') || '';
        appInstance = connection.get('remoteAppInstance') || '';

        // Don't add to the map not yet connected
        if (connection.connecting || !connection.connected) {
          return;
        }

        // Add the hostname to the map
        var host = map[hostName];
        if (!host) {
          host = map[hostName] = {};
        }

        // Add the app to the map
        var app = host[appName];
        if (!app) {
          app = host[appName] = {
            instances: [appInstance],
            probeClasses: connection.get('remoteProbeClasses')
          };
        } else {
          app.instances.push(appInstance);
        }
      });
      log.info('buildMap', map);

      // Set the map if it's changed.  This method is called whenever
      // connections come and go - including firewalled connections which
      // aren't visible in the map.  Only update if the map has changed.
      if (!_.isEqual(map, t.get('map'))) {
        log.info('mapChanged');
        t.set({
          map: map,
          updateSequence: t.updateSequence++
        });
      } else {
        log.info('mapNotChanged');
      }
    }
1248396
private void handleMasterNodeChange() {
    try {
      synchronized (hasActiveServer) {
        if (ZooKeeperUtil.watchAndCheckExists(this, masterZnode)) {
          // A master node exists, there is an active master
          if (LOG.isDebugEnabled()) {
            LOG.debug("A master is now available");
          }
          hasActiveServer.set(true);
        } else {
          // Node is no longer there, cluster does not have an active master
          if (LOG.isDebugEnabled()) {
            LOG.debug("No master available. Notifying waiting threads");
          }
          hasActiveServer.set(false);
          // Notify any thread waiting to become the active master
          hasActiveServer.notifyAll();
        }
      }
    } catch (KeeperException ke) {
      LOG.error("Received an unexpected KeeperException, aborting", ke);
    }
  }
13132
def get_config(name, region=None, key=None, keyid=None, profile=None):
    '''
    
    '''
    conn = _get_conn(region=region, key=key, keyid=keyid, profile=profile)

    if not conn:
        return None
    try:
        cc = conn.describe_cache_clusters(name,
                                          show_cache_node_info=True)
    except boto.exception.BotoServerError as e:
        msg = 'Failed to get config for cache cluster {0}.'.format(name)
        log.error(msg)
        log.debug(e)
        return {}
    cc = cc['DescribeCacheClustersResponse']['DescribeCacheClustersResult']
    cc = cc['CacheClusters'][0]
    ret = odict.OrderedDict()
    attrs = ['engine', 'cache_parameter_group', 'cache_cluster_id',
             'cache_security_groups', 'replication_group_id',
             'auto_minor_version_upgrade', 'num_cache_nodes',
             'preferred_availability_zone', 'security_groups',
             'cache_subnet_group_name', 'engine_version', 'cache_node_type',
             'notification_configuration', 'preferred_maintenance_window',
             'configuration_endpoint', 'cache_cluster_status', 'cache_nodes']
    for key, val in six.iteritems(cc):
        _key = boto.utils.pythonize_name(key)
        if _key not in attrs:
            continue
        if _key == 'cache_parameter_group':
            if val:
                ret[_key] = val['CacheParameterGroupName']
            else:
                ret[_key] = None
        elif _key == 'cache_nodes':
            if val:
                ret[_key] = [k for k in val]
            else:
                ret[_key] = []
        elif _key == 'cache_security_groups':
            if val:
                ret[_key] = [k['CacheSecurityGroupName'] for k in val]
            else:
                ret[_key] = []
        elif _key == 'configuration_endpoint':
            if val:
                ret['port'] = val['Port']
                ret['address'] = val['Address']
            else:
                ret['port'] = None
                ret['address'] = None
        elif _key == 'notification_configuration':
            if val:
                ret['notification_topic_arn'] = val['TopicArn']
            else:
                ret['notification_topic_arn'] = None
        else:
            ret[_key] = val
    return ret
133678
def _is_master_running(self):
        '''
        
        '''
        # Windows doesn't have IPC. Assume the master is running.
        # At worse, it will error 500.
        if salt.utils.platform.is_windows():
            return True

        if self.opts['transport'] == 'tcp':
            ipc_file = 'publish_pull.ipc'
        else:
            ipc_file = 'workers.ipc'
        return os.path.exists(os.path.join(
            self.opts['sock_dir'],
            ipc_file))
289531
def _ask_local_config(self):
        
        options = {"backend": "local", "local-config": {}}

        # Concurrency
        while True:
            concurrency = self._ask_with_default(
                "Maximum concurrency (number of tasks running simultaneously). Leave it empty to use the number of "
                "CPU of your host.", "")
            if concurrency == "":
                break

            try:
                concurrency = int(concurrency)
            except:
                self._display_error("Invalid number")
                continue

            if concurrency <= 0:
                self._display_error("Invalid number")
                continue

            options["local-config"]["concurrency"] = concurrency
            break

        # Debug hostname
        hostname = self._ask_with_default(
            "What is the external hostname/address of your machine? You can leave this empty and let INGInious "
            "autodetect it.", "")
        if hostname != "":
            options["local-config"]["debug_host"] = hostname
        self._display_info(
            "You can now enter the port range for the remote debugging feature of INGInious. Please verify that these "
            "ports are open in your firewall. You can leave this parameters empty, the default is 64100-64200")

        # Debug port range
        port_range = None
        while True:
            start_port = self._ask_with_default("Beginning of the range", "")
            if start_port != "":
                try:
                    start_port = int(start_port)
                except:
                    self._display_error("Invalid number")
                    continue
                end_port = self._ask_with_default("End of the range", str(start_port + 100))
                try:
                    end_port = int(end_port)
                except:
                    self._display_error("Invalid number")
                    continue
                if start_port > end_port:
                    self._display_error("Invalid range")
                    continue
                port_range = str(start_port) + "-" + str(end_port)
            else:
                break
        if port_range != None:
            options["local-config"]["debug_ports"] = port_range

        return options
231969
function (config) {
        if (!config) { return; }

        var replConfig,
            peerNames,

            sanitisePeerConfig,
            source = this.sources[config.identity]; // fn

        // at this stage, the source should not be there
        if (source) {
            source.end(); // in case sources exist (almost unlikely, end them)
            console.log('Warn: duplicate setup of connection found in replication setup for ' + config.identity);
        }

        // set default for configuration objects and also clone them
        // ---------------------------------------------------------

        // clone and set defaults for the configuration variables.
        config = util.clone(config); // clone config. do it here to clone replication config too
        replConfig = util.fill(config.replication || {
            enabled: false // blank config means replication disabled
        }, DEFAULT_REPL_SOURCE_CFG); // extract repl config from main config

        !replConfig.sources && (replConfig.sources = {}); // set default to no source if none specified
        delete config.replication; // we remove the recursive config from main config clone.

        // setup a dumb source as a precaution and since we should fall back to original connection if setup fails
        source = this.sources[config.identity] = db.oneDumbSource();

        // replication explicitly disabled or no peers defined. we do this after ending any ongoing sources
        source._replication = !!replConfig.enabled;
        if (source._replication === false) {
            return;
        }

        // create default connection parameters for cluster
        // ------------------------------------------------

        // get the list of peer names defined in sources and setup the config defaults for peer sources
        peerNames = _.filter(Object.keys(replConfig.sources) || [], function (peer) { // remove all disabled peers
            return (peer = replConfig.sources[peer]) && (peer.enabled !== false);
        });

        sanitisePeerConfig = function (peerConfig) {
            return util.fill(util.extend(peerConfig, {
                database: peerConfig._database || config.database, // force db name to be same
                pool: true,
                waitForConnections: true,
                multipleStatements: true
            }), (replConfig.inheritMaster === true) && config);
        };

        // depending upon the number of peers defined, create simple connection or clustered connection
        // --------------------------------------------------------------------------------------------

        // nothing much to do, if there are no replication source defined
        if (peerNames.length === 0) {
            sails && sails.log.info('smt is re-using master as readonly source');
            // if it is marked to inherit master in replication, we simply create a new source out of the master config
            if (replConfig.inheritMaster === true) {
                this.sources[config.identity] = db.createSource(sanitisePeerConfig({}));
            }
            return;
        }

        // for a single peer, the configuration is simple, we do not need a cluster but a simple pool
        if (peerNames.length === 1) {
            sails && sails.log.info('smt is using 1 "' + config.identity + '" readonly source - ' + peerNames[0]);
            // ensure that the single source's config is taken care of by setting the default and if needed inheriting
            // from master
            this.sources[config.identity] = db.createSource(sanitisePeerConfig(replConfig.sources[peerNames[0]]));
            return;
        }

        // iterate over all sources and normalise and validate their configuration
        util.each(replConfig.sources, sanitisePeerConfig);
        sails && sails.log.info('smt is using ' + peerNames.length + ' "' + config.identity + '" readonly sources - ' +
            peerNames.join(', '));

        // create connections for read-replicas and add it to the peering list
        this.sources[config.identity] = db.createCluster(replConfig);
    }
546759
def discover(self):
        '''
        
        '''
        response = {}
        masters = {}
        self.log.info("Looking for a server discovery")
        self._query()
        self._collect_masters_map(response)
        if not response:
            msg = 'No master has been discovered.'
            self.log.info(msg)
        else:
            for addr, descriptions in response.items():
                for data in descriptions:  # Several masters can run at the same machine.
                    msg = salt.utils.stringutils.to_unicode(data)
                    if msg.startswith(self.signature):
                        msg = msg.split(self.signature)[-1]
                        self.log.debug(
                            "Service announcement at '%s:%s'. Response: '%s'",
                            addr[0], addr[1], msg
                        )
                        if ':E:' in msg:
                            err = msg.split(':E:')[-1]
                            self.log.error(
                                'Error response from the service publisher at %s: %s',
                                addr, err
                            )
                            if "timestamp" in err:
                                self.log.error('Publisher sent shifted timestamp from %s', addr)
                        else:
                            if addr not in masters:
                                masters[addr] = []
                            masters[addr].append(
                                salt.utils.json.loads(msg.split(':@:')[-1], _json_module=_json)
                            )
        return masters
538218
def _read_remote_config(self):
        
        try:
            import requests
        except ImportError:
            requests = None
        if not requests:
            raise ValueError(
                'Remote config URL specified but requests not installed')
        result = requests.get(self._file_path)
        if not result.ok:
            raise ValueError(
                'Failed to retrieve remote config: {}'.format(
                    result.status_code))
        return result.text
1077558
def get_master(self, host, port, sentinel_port, sentinel_name):
        
        if sentinel_port and sentinel_name:
            master = Sentinel([(host, sentinel_port)], socket_timeout=1)\
                .discover_master(sentinel_name)
            return master
        return host, port
671048
def _handle_tag_master_connected(self, tag, data):
        '''
        
        '''
        # handle this event only once. otherwise it will pollute the log
        # also if master type is failover all the reconnection work is done
        # by `disconnected` event handler and this event must never happen,
        # anyway check it to be sure
        if not self.connected and self.opts['master_type'] != 'failover':
            log.info('Connection to master %s re-established', self.opts['master'])
            self.connected = True
            # modify the __master_alive job to only fire,
            # if the connection is lost again
            if self.opts['transport'] != 'tcp':
                if self.opts['master_alive_interval'] > 0:
                    schedule = {
                       'function': 'status.master',
                       'seconds': self.opts['master_alive_interval'],
                       'jid_include': True,
                       'maxrunning': 1,
                       'return_job': False,
                       'kwargs': {'master': self.opts['master'],
                                   'connected': True}
                    }

                    self.schedule.modify_job(name=master_event(type='alive', master=self.opts['master']),
                                             schedule=schedule)
                else:
                    self.schedule.delete_job(name=master_event(type='alive', master=self.opts['master']), persist=True)
1249822
def autoconf(self):
        
        serverInfo = MemcachedInfo(self._host,  self._port, self._socket_file)
        return (serverInfo is not None)
429554
async def discover_master(self, service, timeout):
        """"""
        # TODO: get lock
        idle_timeout = timeout
        # FIXME: single timeout used 4 times;
        #   meaning discovery can take up to:
        #   3 * timeout * (sentinels count)
        #
        #   having one global timeout also can leed to
        #   a problem when not all sentinels are checked.

        # use a copy, cause pools can change
        pools = self._pools[:]
        for sentinel in pools:
            try:
                with async_timeout(timeout, loop=self._loop):
                    address = await self._get_masters_address(
                        sentinel, service)

                pool = self._masters[service]
                with async_timeout(timeout, loop=self._loop), \
                        contextlib.ExitStack() as stack:
                    conn = await pool._create_new_connection(address)
                    stack.callback(conn.close)
                    await self._verify_service_role(conn, 'master')
                    stack.pop_all()

                return conn
            except asyncio.CancelledError:
                # we must correctly handle CancelledError(s):
                #   application may be stopped or function can be cancelled
                #   by outer timeout, so we must stop the look up.
                raise
            except asyncio.TimeoutError:
                continue
            except DiscoverError as err:
                sentinel_logger.debug("DiscoverError(%r, %s): %r",
                                      sentinel, service, err)
                await asyncio.sleep(idle_timeout, loop=self._loop)
                continue
            except RedisError as err:
                raise MasterReplyError("Service {} error".format(service), err)
            except Exception:
                # TODO: clear (drop) connections to schedule reconnect
                await asyncio.sleep(idle_timeout, loop=self._loop)
                continue
        else:
            raise MasterNotFoundError("No master found for {}".format(service))
628178
def get_head_node_ip(config_file, override_cluster_name):
    """"""

    config = yaml.load(open(config_file).read())
    if override_cluster_name is not None:
        config["cluster_name"] = override_cluster_name

    provider = get_node_provider(config["provider"], config["cluster_name"])
    try:
        head_node = _get_head_node(config, config_file, override_cluster_name)
        if config.get("provider", {}).get("use_internal_ips", False) is True:
            head_node_ip = provider.internal_ip(head_node)
        else:
            head_node_ip = provider.external_ip(head_node)
    finally:
        provider.cleanup()

    return head_node_ip
67798
def do_GET(self, write_status_code_only=False):
        """"""

        path = '/master' if self.path == '/' else self.path
        response = self.get_postgresql_status()

        patroni = self.server.patroni
        cluster = patroni.dcs.cluster

        if not cluster and patroni.ha.is_paused():
            primary_status_code = 200 if response['role'] == 'master' else 503
        else:
            primary_status_code = 200 if patroni.ha.is_leader() else 503

        replica_status_code = 200 if not patroni.noloadbalance and response.get('role') == 'replica' else 503
        status_code = 503

        if patroni.ha.is_standby_cluster() and ('standby_leader' in path or 'standby-leader' in path):
            status_code = 200 if patroni.ha.is_leader() else 503
        elif 'master' in path or 'leader' in path or 'primary' in path or 'read-write' in path:
            status_code = primary_status_code
        elif 'replica' in path:
            status_code = replica_status_code
        elif 'read-only' in path:
            status_code = 200 if primary_status_code == 200 else replica_status_code
        elif cluster:  # dcs is available
            is_synchronous = cluster.is_synchronous_mode() and cluster.sync \
                    and cluster.sync.sync_standby == patroni.postgresql.name
            if path in ('/sync', '/synchronous') and is_synchronous:
                status_code = replica_status_code
            elif path in ('/async', '/asynchronous') and not is_synchronous:
                status_code = replica_status_code

        if write_status_code_only:  # when haproxy sends OPTIONS request it reads only status code and nothing more
            message = self.responses[status_code][0]
            self.wfile.write('{0} {1} {2}\r\n'.format(self.protocol_version, status_code, message).encode('utf-8'))
        else:
            self._write_status_response(status_code, response)
887361
def client_socket(self, config):
		
		if self.__client_socket is None:
			self.__client_socket = self.create_client_socket(config)
		return self.__client_socket