我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用charmhelpers.core.hookenv.unit_get()。
def write_prometheus_config_def(): config = hookenv.config() port = config.get('port', '9090') check_ports(port) if config.get('external_url', False): vars = { 'private_address': hookenv.unit_get('private-address'), 'public_address': hookenv.unit_get('public-address'), # prometheus default: 'port': port, } runtime_args('-web.external-url', config['external_url'].format(**vars)) args = runtime_args() hookenv.log('runtime_args: {}'.format(args)) if args: render(source=PROMETHEUS_DEF_TMPL, target=PROMETHEUS_DEF, context={'args': args}, ) set_state('prometheus.do-restart') remove_state('prometheus.do-reconfig-def')
def provide_data(self): return { 'host': hookenv.unit_get('private-address'), 'port': 80, }
def sniff_iface(f): """Ensure decorated function is called with a value for iface. If no iface provided, inject net iface inferred from unit private address. """ def iface_sniffer(*args, **kwargs): if not kwargs.get('iface', None): kwargs['iface'] = get_iface_from_addr(unit_get('private-address')) return f(*args, **kwargs) return iface_sniffer
def get_local_ip() -> Result: """ Returns the local IPAddr address associated with this server # Failures Returns a GlusterError representing any failure that may have happened while trying to query this information. """ ip_addr = get_host_ip(unit_get('private-address')) try: parsed = ip_address(address=ip_addr) return Ok(parsed) # Resolves a str hostname into a ip address. except ValueError: return Err("failed to parse ip address: {}".format(ip_addr))
def get_database_setup(self): """Provide the default database credentials as a list of 3-tuples returns a structure of: [ {'database': <database>, 'username': <username>, 'hostname': <hostname of this unit> 'prefix': <the optional prefix for the database>, }, ] :returns [{'database': ...}, ...]: credentials for multiple databases """ host = None try: host = hookenv.network_get_primary_address('shared-db') except NotImplementedError: host = hookenv.unit_get('private-address') return [ dict( database=self.config['database'], username=self.config['database-user'], hostname=host, ) ] # Determine the charm class by the supported release
def write_prometheus_config_yml(): config = hookenv.config() target_jobs = unitdata.kv().get('target_jobs', []) scrape_jobs = unitdata.kv().get('scrape_jobs', []) # transform eg. 'h1:p1 , h2:p2' (string), to ['h1:p1', 'h2:p2'] (list) static_targets = None if config.get('static-targets'): static_targets = [x.strip() for x in config.get('static-targets', '').split(',')] default_monitor_name = '{}-monitor'.format(hookenv.service_name()) options = { 'scrape_interval': config['scrape-interval'], 'evaluation_interval': config['evaluation-interval'], 'static_targets': static_targets, 'private_address': hookenv.unit_get('private-address'), 'monitor_name': config.get('monitor_name', default_monitor_name), 'jobs': target_jobs, 'scrape_jobs': scrape_jobs, } # custom-rules content must be passed verbatim with e.g. # juju set prometheus custom-rules @my.rules if config.get('custom-rules'): custom_rules = config['custom-rules'] with open(CUSTOM_RULES_PATH, 'w') as fh: fh.write(custom_rules) options['custom_rules_file'] = CUSTOM_RULES_PATH render(source=PROMETHEUS_YML_TMPL, target=PROMETHEUS_YML, context=options ) validate_config() set_state('prometheus.do-restart') remove_state('prometheus.do-reconfig-yml')