我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用charmhelpers.core.host.init_is_systemd()。
def install_ntpmon(): """ Install package dependencies, source files, and startup configuration. """ hookenv.log('installing ntpmon dependencies') apt_install(['python3-psutil']) hookenv.log('installing ntpmon') host.rsync('src/', ntpmon_dir) if host.init_is_systemd(): hookenv.log('installing ntpmon systemd configuration') host.rsync('src/' + service_name + '.systemd', systemd_config) subprocess.call(['systemd', 'daemon-reload']) else: hookenv.log('installing ntpmon upstart configuration') host.rsync('src/' + service_name + '.upstart', upstart_config) set_state('ntpmon.installed') remove_state('ntpmon.configured')
def register_configs(): """ Register config files with their respective contexts. Regstration of some configs may not be required depending on existing of certain relations. """ # if called without anything installed (eg during install hook) # just default to earliest supported release. configs dont get touched # till post-install, anyway. release = (get_os_codename_package('ceilometer-common', fatal=False) or 'grizzly') configs = templating.OSConfigRenderer(templates_dir=TEMPLATES, openstack_release=release) for conf in (CEILOMETER_CONF, HAPROXY_CONF): configs.register(conf, CONFIG_FILES[conf]['hook_contexts']) if init_is_systemd(): configs.register( CEILOMETER_API_SYSTEMD_CONF, CONFIG_FILES[CEILOMETER_API_SYSTEMD_CONF]['hook_contexts'] ) if os.path.exists('/etc/apache2/conf-available'): configs.register(HTTPS_APACHE_24_CONF, CONFIG_FILES[HTTPS_APACHE_24_CONF]['hook_contexts']) else: configs.register(HTTPS_APACHE_CONF, CONFIG_FILES[HTTPS_APACHE_CONF]['hook_contexts']) if enable_memcache(release=release): configs.register(MEMCACHED_CONF, [context.MemcacheContext()]) if run_in_apache(): wsgi_script = "/usr/share/ceilometer/app.wsgi" configs.register(WSGI_CEILOMETER_API_CONF, [context.WSGIWorkerConfigContext(name="ceilometer", script=wsgi_script), CeilometerContext(), HAProxyContext()]) return configs
def reload_systemd(): """Reload systemd configuration on systemd based installs """ if init_is_systemd(): subprocess.check_call(['systemctl', 'daemon-reload'])
def reload_and_restart(): if ch_host.init_is_systemd(): subprocess.check_call(['systemctl', 'daemon-reload']) ch_host.service_restart('aodh-api')
def reload_and_restart(): if ch_host.init_is_systemd(): subprocess.check_call(['systemctl', 'daemon-reload']) # no need to restart aodh-api in ocata and onwards
def test_init_is_systemd_upstart(self, path, lsb_release): """Upstart based init is correctly detected""" lsb_release.return_value = {'DISTRIB_CODENAME': 'whatever'} path.isdir.return_value = False self.assertFalse(host.init_is_systemd()) path.isdir.assert_called_with('/run/systemd/system')
def test_init_is_systemd_system(self, path, lsb_release): """Systemd based init is correctly detected""" lsb_release.return_value = {'DISTRIB_CODENAME': 'whatever'} path.isdir.return_value = True self.assertTrue(host.init_is_systemd()) path.isdir.assert_called_with('/run/systemd/system')
def test_init_is_systemd_trusty(self, path, lsb_release): # Never returns true under trusty, even if the systemd # packages have been installed. lp:1670944 lsb_release.return_value = {'DISTRIB_CODENAME': 'trusty'} path.isdir.return_value = True self.assertFalse(host.init_is_systemd()) self.assertFalse(path.isdir.called)
def test_is_container_with_systemd_container(self, call, init_is_systemd, mock_os): init_is_systemd.return_value = True call.return_value = 0 self.assertTrue(host.is_container()) call.assert_called_with(['systemd-detect-virt', '--container'])
def test_is_container_with_systemd_non_container(self, call, init_is_systemd, mock_os): init_is_systemd.return_value = True call.return_value = 1 self.assertFalse(host.is_container()) call.assert_called_with(['systemd-detect-virt', '--container'])
def test_is_container_with_upstart_not_container(self, call, init_is_systemd, mock_os): init_is_systemd.return_value = False mock_os.path.exists.return_value = False self.assertFalse(host.is_container()) mock_os.path.exists.assert_called_with('/run/container_type')
def add_init_service_checks(nrpe, services, unit_name, immediate_check=True): """ Add checks for each service in list :param NRPE nrpe: NRPE object to add check to :param list services: List of services to check :param str unit_name: Unit name to use in check description :param bool immediate_check: For sysv init, run the service check immediately """ for svc in services: # Don't add a check for these services from neutron-gateway if svc in ['ext-port', 'os-charm-phy-nic-mtu']: next upstart_init = '/etc/init/%s.conf' % svc sysv_init = '/etc/init.d/%s' % svc if host.init_is_systemd(): nrpe.add_check( shortname=svc, description='process check {%s}' % unit_name, check_cmd='check_systemd.py %s' % svc ) elif os.path.exists(upstart_init): nrpe.add_check( shortname=svc, description='process check {%s}' % unit_name, check_cmd='check_upstart_job %s' % svc ) elif os.path.exists(sysv_init): cronpath = '/etc/cron.d/nagios-service-check-%s' % svc checkpath = '%s/service-check-%s.txt' % (nrpe.homedir, svc) croncmd = ( '/usr/local/lib/nagios/plugins/check_exit_status.pl ' '-e -s /etc/init.d/%s status' % svc ) cron_file = '*/5 * * * * root %s > %s\n' % (croncmd, checkpath) f = open(cronpath, 'w') f.write(cron_file) f.close() nrpe.add_check( shortname=svc, description='service check {%s}' % unit_name, check_cmd='check_status_file.py -f %s' % checkpath, ) # if /var/lib/nagios doesn't exist open(checkpath, 'w') will fail # (LP: #1670223). if immediate_check and os.path.isdir(nrpe.homedir): f = open(checkpath, 'w') subprocess.call( croncmd.split(), stdout=f, stderr=subprocess.STDOUT ) f.close() os.chmod(checkpath, 0o644)
