Python charmhelpers.core.hookenv 模块，WARNING 实例源码

def service_ports(self):
        """Dict of service names and the ports they listen on

        @return {'svc1': ['portA', 'portB'], 'svc2': ['portC', 'portD'], ...}
        """
        # Note(AJK) - ensure that service ports is always in the same order
        service_ports = collections.OrderedDict()
        if self.port_map:
            for service in sorted(self.port_map.keys()):
                port_types = sorted(list(self.port_map[service].keys()))
                for port_type in port_types:
                    listen_port = self.port_map[service][port_type]
                    key = '{}_{}'.format(service, port_type)
                    used_ports = [v[0] for v in service_ports.values()]
                    if listen_port in used_ports:
                        hookenv.log("Not adding haproxy listen stanza for {} "
                                    "port is already in use".format(key),
                                    level=hookenv.WARNING)
                        continue
                    service_ports[key] = [
                        self.port_map[service][port_type],
                        ch_cluster.determine_apache_port(
                            self.port_map[service][port_type],
                            singlenode_mode=True)]

        return service_ports

def ensure_api_responding(cls):
        """Check that the api service is responding.

        The retry_on_exception decorator will cause this method to be called
        until it succeeds or retry limit is exceeded"""
        hookenv.log('Checking API service is responding',
                    level=hookenv.WARNING)
        check_cmd = ['reactive/designate_utils.py', 'server-list']
        subprocess.check_call(check_cmd)

def create_initial_servers_and_domains(cls):
        """Create the nameserver entry and domains based on the charm user
        supplied config

        NOTE(AJK): This only wants to be done ONCE and by the leader, so we use
        leader settings to store that we've done it, after it's successfully
        completed.

        @returns None
        """
        KEY = 'create_initial_servers_and_domains'
        if hookenv.is_leader() and not hookenv.leader_get(KEY):
            nova_domain_name = hookenv.config('nova-domain')
            neutron_domain_name = hookenv.config('neutron-domain')
            with cls.check_zone_ids(nova_domain_name, neutron_domain_name):
                if hookenv.config('nameservers'):
                    for ns in hookenv.config('nameservers').split():
                        cls.create_server(ns)
                else:
                    hookenv.log('No nameserver specified, skipping creation of'
                                'nova and neutron domains',
                                level=hookenv.WARNING)
                    return
                if nova_domain_name:
                    cls.create_domain(
                        nova_domain_name,
                        hookenv.config('nova-domain-email'))
                if neutron_domain_name:
                    cls.create_domain(
                        neutron_domain_name,
                        hookenv.config('neutron-domain-email'))
            # if this fails, we weren't the leader any more; another unit may
            # attempt to do this too.
            hookenv.leader_set({KEY: 'done'})

def retrieve_zones(self, cluster_relation=None):
        """Retrieve and install zones file

        Check if published sync target was created after this units sync
        request was sent, if it was install the zones file. Alternatively if
        no peer relation was set then assume the current sync target is to be
        used regardless of when it was created.

        :param cluster_relation: OpenstackHAPeers() interface class
        :returns: None
        """

        request_time = None
        if cluster_relation:
            request_times = list(set(
                cluster_relation.retrieve_local(CLUSTER_SYNC_KEY)))
            request_time = request_times[0]
        sync_time = DesignateBindCharm.get_sync_time()
        if request_time and request_time > sync_time:
            hookenv.log(('Request for sync sent but remote sync time is too'
                         ' old, defering until a more up-to-date target is '
                         'available'),
                        level=hookenv.WARNING)
        else:
            self.service_control('stop', ['bind9'])
            url = DesignateBindCharm.get_sync_src()
            self.wget_file(url, ZONE_DIR)
            tar_file = url.split('/')[-1]
            subprocess.check_call(['tar', 'xf', tar_file], cwd=ZONE_DIR)
            os.remove('{}/{}'.format(ZONE_DIR, tar_file))
            self.service_control('start', ['bind9'])
            reactive.remove_state('sync.request.sent')
            reactive.set_state('zones.initialised')

def test_logs_messages_with_alternative_levels(self, mock_call):
        alternative_levels = [
            hookenv.CRITICAL,
            hookenv.ERROR,
            hookenv.WARNING,
            hookenv.INFO,
        ]

        for level in alternative_levels:
            hookenv.log('foo', level)
            mock_call.assert_called_with(['juju-log', '-l', level, 'foo'])

def __init__(self, port_map=None, service_name=None, charm_instance=None):
        """
        Note passing port_map and service_name is deprecated, but supported for
        backwards compatibility.  The port_map and service_name can be obtained
        from the self.charm_instance weak reference.
        :param  port_map: Map containing service names and the ports used e.g.
                port_map = {
                        'svc1': {
                        'admin': 9001,
                        'public': 9001,
                        'int': 9001,
                    },
                        'svc2': {
                        'admin': 9002,
                        'public': 9002,
                        'int': 9002,
                    },
                }
        :param service_name: Name of service being deployed
        :param charm_instance: a charm instance that will be passed to the base
            constructor
        """
        super(APIConfigurationAdapter, self).__init__(
            charm_instance=charm_instance)
        if port_map is not None:
            hookenv.log(
                "DEPRECATION: should not use port_map parameter in "
                "APIConfigurationAdapter.__init__()", level=hookenv.WARNING)
            self.port_map = port_map
        elif self.charm_instance is not None:
            self.port_map = self.charm_instance.api_ports
        else:
            self.port_map = None
        if service_name is not None:
            hookenv.log(
                "DEPRECATION: should not use service_name parameter in "
                "APIConfigurationAdapter.__init__()", level=hookenv.WARNING)
            self.service_name = service_name
        elif self.charm_instance is not None:
            self.service_name = self.charm_instance.name
        else:
            self.service_name = None
        self.__network_addresses = None

def __init__(self, relations, options=None, options_instance=None,
                 charm_instance=None):
        """
        :param relations: List of instances of relation classes
        :param options: Configuration class to use (DEPRECATED)
        :param options_instance: Instance of Configuration class to use
        :param charm_instance: optional charm_instance that is captured as a
            weakref for use on the adapter.
        """
        self._charm_instance_weakref = None
        if charm_instance is not None:
            self._charm_instance_weakref = weakref.ref(charm_instance)
        self._relations = set()
        if options is not None:
            hookenv.log("The 'options' argument is deprecated please use "
                        "options_instance instead.", level=hookenv.WARNING)
            self.options = options()
        elif options_instance is not None:
            self.options = options_instance
        else:
            # create a default, customised ConfigurationAdapter if the
            # APIConfigurationAdapter is needed as a base, then it must be
            # passed as an instance on the options_instance First pull the
            # configuration class from the charm instance (if it's available).
            base_cls = None
            if self.charm_instance:
                base_cls = getattr(self.charm_instance, 'configuration_class',
                                   base_cls)
            self.options = make_default_options(base_cls, self.charm_instance)
        self._relations.add('options')
        # walk the mro() from object to this class to build up the _adapters
        # ensure that all of the relations' have their '-' turned into a '_' to
        # ensure that everything is consistent in the class.
        self._adapters = {}
        for cls in reversed(self.__class__.mro()):
            self._adapters.update(
                {k.replace('-', '_'): v
                 for k, v in getattr(cls, 'relation_adapters', {}).items()})
        # now we have to add in any customisations to those adapters
        for relation, properties in _custom_adapter_properties.items():
            relation = relation.replace('-', '_')
            try:
                cls = self._adapters[relation]
            except KeyError:
                cls = OpenStackRelationAdapter
            self._adapters[relation] = make_default_relation_adapter(
                cls, relation, properties)
        self.add_relations(relations)

def set_authentication_data(self, value, name=None):
        """Set the authentication data to the plugin charm.  This is to enable
        the plugin to either 'talk' to OpenStack or to provide authentication
        data into the configuraiton sections that it needs to set (the generic
        backend needs to do this).

        The authentication data format is:
        {
            'username': <value>
            'password': <value>
            'project_domain_id': <value>
            'project_name': <value>
            'user_domain_id': <value>
            'auth_uri': <value>
            'auth_url': <value>
            'auth_type': <value>  # 'password', typically
        }

        :param value: a dictionary of data to set.
        :param name: OPTIONAL - target the config at a particular name only
        """
        keys = {'username', 'password', 'project_domain_id', 'project_name',
                'user_domain_id', 'auth_uri', 'auth_url', 'auth_type'}
        passed_keys = set(value.keys())
        if passed_keys.difference(keys) or keys.difference(passed_keys):
            hookenv.log(
                "Setting Authentication data; there may be missing or mispelt "
                "keys: passed: {}".format(passed_keys),
                level=hookenv.WARNING)
        # need to check for each conversation whether we've sent the data, or
        # whether it is different, and then set the local & remote only if that
        # is the case.
        for conversation in self.conversations():
            if conversation.scope is None:
                # the conversation has gone away; ignore it
                continue
            if name is not None:
                conversation_name = self.get_remote('_name', default=None,
                                                    scope=conversation.scope)
                if name != conversation_name:
                    continue
            existing_auth_data = self.get_local('_authentication_data',
                                                default=None,
                                                scope=conversation.scope)
            if existing_auth_data is not None:
                # see if they are different
                existing_auth = json.loads(existing_auth_data)["data"]
                if (existing_auth.keys() == value.keys() and
                        all([v == value[k]
                             for k, v in existing_auth.items()])):
                    # the values haven't changed, so don't set them again
                    continue
            self.set_local(_authentication_data=json.dumps({"data": value}),
                           scope=conversation.scope)
            self.set_remote(_authentication_data=json.dumps({"data": value}),
                            scope=conversation.scope)