我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用charmhelpers.core.host.service()。
def write(self, nagios_context, hostname, nagios_servicegroups): nrpe_check_file = self._get_check_filename() with open(nrpe_check_file, 'w') as nrpe_check_config: nrpe_check_config.write("# check {}\n".format(self.shortname)) if nagios_servicegroups: nrpe_check_config.write( "# The following header was added automatically by juju\n") nrpe_check_config.write( "# Modifying it will affect nagios monitoring and alerting\n") nrpe_check_config.write( "# servicegroups: {}\n".format(nagios_servicegroups)) nrpe_check_config.write("command[{}]={}\n".format( self.command, self.check_cmd)) if not os.path.exists(NRPE.nagios_exportdir): log('Not writing service config as {} is not accessible'.format( NRPE.nagios_exportdir)) else: self.write_service_config(nagios_context, hostname, nagios_servicegroups)
def test_pauses_a_running_upstart_service(self, service, systemd, service_running): """Pause on a running service will call service stop.""" service_name = 'foo-service' service.side_effect = [True] systemd.return_value = False service_running.return_value = True tempdir = mkdtemp(prefix="test_pauses_an_upstart_service") conf_path = os.path.join(tempdir, "{}.conf".format(service_name)) # Just needs to exist with open(conf_path, "w") as fh: fh.write("") self.addCleanup(rmtree, tempdir) self.assertTrue(host.service_pause(service_name, init_dir=tempdir)) service.assert_called_with('stop', service_name) override_path = os.path.join( tempdir, "{}.override".format(service_name)) with open(override_path, "r") as fh: override_contents = fh.read() self.assertEqual("manual\n", override_contents)
def test_pauses_a_stopped_upstart_service(self, service, systemd, service_running): """Pause on a stopped service will not call service stop.""" service_name = 'foo-service' service.side_effect = [True] systemd.return_value = False service_running.return_value = False tempdir = mkdtemp(prefix="test_pauses_an_upstart_service") conf_path = os.path.join(tempdir, "{}.conf".format(service_name)) # Just needs to exist with open(conf_path, "w") as fh: fh.write("") self.addCleanup(rmtree, tempdir) self.assertTrue(host.service_pause(service_name, init_dir=tempdir)) # Stop isn't called because service is already stopped self.assertRaises( AssertionError, service.assert_called_with, 'stop', service_name) override_path = os.path.join( tempdir, "{}.override".format(service_name)) with open(override_path, "r") as fh: override_contents = fh.read() self.assertEqual("manual\n", override_contents)
def test_pauses_a_stopped_sysv_service(self, service, check_call, systemd, service_running): """Pause does not call service stop on a stopped sysv service.""" service_name = 'foo-service' service.side_effect = [True] systemd.return_value = False service_running.return_value = False tempdir = mkdtemp(prefix="test_pauses_a_sysv_service") sysv_path = os.path.join(tempdir, service_name) # Just needs to exist with open(sysv_path, "w") as fh: fh.write("") self.addCleanup(rmtree, tempdir) self.assertTrue(host.service_pause( service_name, init_dir=tempdir, initd_dir=tempdir)) # Stop isn't called because service is already stopped self.assertRaises( AssertionError, service.assert_called_with, 'stop', service_name) check_call.assert_called_with(["update-rc.d", service_name, "disable"])
def __init__(self, shortname, description, check_cmd): super(Check, self).__init__() # XXX: could be better to calculate this from the service name if not re.match(self.shortname_re, shortname): raise CheckException("shortname must match {}".format( Check.shortname_re)) self.shortname = shortname self.command = "check_{}".format(shortname) # Note: a set of invalid characters is defined by the # Nagios server config # The default is: illegal_object_name_chars=`~!$%^&*"|'<>?,()= self.description = description self.check_cmd = self._locate_cmd(check_cmd)
def write(self): try: nagios_uid = pwd.getpwnam('nagios').pw_uid nagios_gid = grp.getgrnam('nagios').gr_gid except Exception: log("Nagios user not set up, nrpe checks not updated") return if not os.path.exists(NRPE.nagios_logdir): os.mkdir(NRPE.nagios_logdir) os.chown(NRPE.nagios_logdir, nagios_uid, nagios_gid) nrpe_monitors = {} monitors = {"monitors": {"remote": {"nrpe": nrpe_monitors}}} for nrpecheck in self.checks: nrpecheck.write(self.nagios_context, self.hostname, self.nagios_servicegroups) nrpe_monitors[nrpecheck.shortname] = { "command": nrpecheck.command, } # update-status hooks are configured to firing every 5 minutes by # default. When nagios-nrpe-server is restarted, the nagios server # reports checks failing causing unneccessary alerts. Let's not restart # on update-status hooks. if not hook_name() == 'update-status': service('restart', 'nagios-nrpe-server') monitor_ids = relation_ids("local-monitors") + \ relation_ids("nrpe-external-master") for rid in monitor_ids: relation_set(relation_id=rid, monitors=yaml.dump(monitors))
def add_haproxy_checks(nrpe, unit_name): """ Add checks for each service in list :param NRPE nrpe: NRPE object to add check to :param str unit_name: Unit name to use in check description """ nrpe.add_check( shortname='haproxy_servers', description='Check HAProxy {%s}' % unit_name, check_cmd='check_haproxy.sh') nrpe.add_check( shortname='haproxy_queue', description='Check HAProxy queue depth {%s}' % unit_name, check_cmd='check_haproxy_queue_depth.sh')
def service(subparser): subparser.add_argument("action", help="The action to perform (start, stop, etc...)") subparser.add_argument("service_name", help="Name of the service to control") return host.service
