# -*- coding: utf-8 -*-
import centreonapi.webservice.configuration.common as common
import centreonapi.webservice.configuration.factory.pollerfactory as pollerfactory
import centreonapi.webservice.configuration.factory.hostfactory as hostfactory
import centreonapi.webservice.configuration.factory.hostgroupfactory as hostgroupfactory
import centreonapi.webservice.configuration.factory.contactgroupfactory as contactgroupfactory
import centreonapi.webservice.configuration.factory.contactfactory as contactfactory
from centreonapi.webservice.configuration.hostgroups import HostGroup
from centreonapi.webservice.configuration.contact import ContactGroup, Contact
from centreonapi.webservice import Webservice
[docs]class Host(hostfactory.ObjHost):
def __init__(self, properties):
#super(Host, self).__init__()
self.webservice = Webservice.getInstance()
self.__clapi_action = 'HOST'
self.id = properties.get('id')
self.name = properties.get('name')
self.activate = properties.get('activate')
self.address = properties.get('address')
self.alias = properties.get('alias')
self.macros = {}
self.templates = {}
self.parents = {}
self.hostgroups = {}
self.contactgroups = {}
self.contacts = {}
self.params = {}
self.state = properties.get('state')
[docs] def getmacro(self):
"""
Get host's macros.
:return:
- bool: API call status
- dict: all macros on hosts
"""
self.macros.clear()
state, macro = self.webservice.call_clapi(
'getmacro',
self.__clapi_action,
self.name)
if state:
if len(macro['result']) > 0:
for m in macro['result']:
macro_obj = HostMacro(m)
self.macros[macro_obj.engine_name] = macro_obj
return state, self.macros
else:
return state, None
else:
return state, macro
[docs] def setmacro(self, name, value, is_password=None, description=None):
"""
Add or update macro
:param name: macro name
:param value: macro value
:param is_password: set to 1 if a macro password
:param description: macro description
:return:
"""
if description is None:
description = ''
if is_password is None:
is_password = 0
macro = name.replace("_HOST", "").replace("$", "").upper()
values = [self.name, macro, value, is_password, description]
return self.webservice.call_clapi(
'setmacro',
self.__clapi_action,
values)
[docs] def deletemacro(self, macro):
"""
Remove macro
:param macro: HostMacro
:return:
"""
values = [self.name,
"|".join(hostfactory.build_param_hostmacro(macro))]
return self.webservice.call_clapi(
'delmacro',
self.__clapi_action,
values)
[docs] def gettemplate(self):
"""
Get host's Template
:return:
- bool: API call status
- dict: all templates on hosts
"""
state, template = self.webservice.call_clapi(
'gettemplate',
self.__clapi_action,
self.name)
if state:
if len(template['result']) > 0:
for t in template['result']:
template_obj = HostTemplate(t)
self.templates[template_obj.name] = template_obj
return state, self.templates
else:
return state, None
else:
return state, template
[docs] def settemplate(self, template=None):
"""
Define HostTemplate on host
:param template: HostTemplate, list(HosTemplate), str
:return:
"""
values = [self.name,
"|".join(hostfactory.build_param_hosttemplate(template))]
return self.webservice.call_clapi(
'settemplate',
self.__clapi_action,
values)
[docs] def addtemplate(self, template=None):
"""
Add new HostTemplate on host
:param template: HostTemplate, list(HosTemplate), str
:return:
"""
values = [self.name,
"|".join(hostfactory.build_param_hosttemplate(template))]
return self.webservice.call_clapi(
'addtemplate',
self.__clapi_action,
values)
[docs] def deletetemplate(self, template=None):
"""
Delete HostTemplate on host
:param template: HostTemplate, list(HosTemplate), str
:return:
"""
values = [self.name,
str("|".join(hostfactory.build_param_hosttemplate(template)))]
return self.webservice.call_clapi(
'deltemplate',
self.__clapi_action,
values)
[docs] def applytemplate(self):
return self.webservice.call_clapi(
'applytpl',
self.__clapi_action,
self.name)
[docs] def enable(self):
s, e = self.webservice.call_clapi(
'enable',
self.__clapi_action,
self.name)
if s:
self.activate = "1"
return s, e
[docs] def disable(self):
s, e = self.webservice.call_clapi(
'disable',
self.__clapi_action,
self.name)
if s:
self.activate = "0"
return s, e
[docs] def setinstance(self, instance):
values = [self.name,
str(pollerfactory.build_param_poller(instance)[0])]
return self.webservice.call_clapi(
'setinstance',
self.__clapi_action,
values)
[docs] def status(self):
values = {'search': self.name}
self.state = self.webservice.centreon_realtime(
'list',
'hosts',
values)[0]['state']
return self.state
[docs] def getparent(self):
state, parent = self.webservice.call_clapi(
'getparent',
self.__clapi_action,
self.name)
if state:
if len(parent['result']) > 0:
for p in parent['result']:
parent_obj = Host(p)
self.parents[parent_obj.name] = parent_obj
return state, self.parents
return state, None
else:
return state, parent
[docs] def addparent(self, parents):
values = [self.name,
"|".join(hostfactory.build_param_host(parents))]
return self.webservice.call_clapi(
'addparent',
self.__clapi_action,
values)
[docs] def setparent(self, parents):
values = [self.name,
"|".join(hostfactory.build_param_host(parents))]
return self.webservice.call_clapi(
'setparent',
self.__clapi_action,
values)
[docs] def deleteparent(self, parents):
values = [self.name,
"|".join(hostfactory.build_param_host(parents))]
return self.webservice.call_clapi(
'delparent',
self.__clapi_action,
values)
[docs] def gethostgroup(self):
state, hgs = self.webservice.call_clapi(
'gethostgroup',
self.__clapi_action,
self.name)
if state:
if len(hgs['result']) > 0:
for h in hgs['result']:
hg_obj = HostGroup(h)
self.hostgroups[hg_obj.name] = hg_obj
return state, self.hostgroups
else:
return state, None
else:
return state, hgs
[docs] def addhostgroup(self, hostgroup=None):
values = [self.name,
"|".join(hostgroupfactory.build_param_hostgroup(hostgroup))]
return self.webservice.call_clapi(
'addhostgroup',
self.__clapi_action,
values)
[docs] def sethostgroup(self, hostgroup=None):
values = [self.name,
"|".join(hostgroupfactory.build_param_hostgroup(hostgroup))]
return self.webservice.call_clapi(
'sethostgroup',
self.__clapi_action,
values)
[docs] def deletehostgroup(self, hostgroup=None):
values = [self.name,
"|".join(hostgroupfactory.build_param_hostgroup(hostgroup))]
return self.webservice.call_clapi(
'delhostgroup',
self.__clapi_action,
values)
[docs] def setseverity(self, name):
pass
#return self.webservice.call_clapi(
# 'setseverity',
# self.__clapi_action,
# [self.name, name])
[docs] def unsetseverity(self):
pass
#return self.webservice.call_clapi(
# 'unsetseverity',
# self.__clapi_action,
# self.name)
[docs] def setparam(self, name, value):
values = [self.name, name, value]
s, param = self.webservice.call_clapi(
'setparam',
self.__clapi_action,
values)
if s:
self.params[name] = value
return s, param
[docs] def getparam(self, name):
values = [self.name, 'activate|' + "|".join(name)]
s, param = self.webservice.call_clapi(
'getparam',
self.__clapi_action,
values
)
if s:
current_param = {}
for p in param['result']:
current_param[p.split(":", 1)[0].strip()] = p.split(":", 1)[1].strip()
self.params[p.split(":", 1)[0].strip()] = p.split(":", 1)[1].strip()
return s, current_param
else:
return s, param
[docs] def getparams(self):
centreon_host_params = ['2d_coords', '3d_coords', 'action_url', 'activate', 'active_checks_enabled',
'address', 'alias', 'check_command', 'check_command_arguments', 'check_interval',
'check_freshness', 'check_period', 'contact_additive_inheritance',
'cg_additive_inheritance', 'event_handler', 'event_handler_arguments',
'event_handler_enabled', 'first_notification_delay', 'flap_detection_enabled',
'flap_detection_options', 'host_high_flap_threshold', 'host_low_flap_threshold',
'icon_image', 'icon_image_alt', 'max_check_attempts', 'name', 'notes', 'notes_url',
'notifications_enabled', 'notification_interval', 'notification_options',
'notification_period', 'recovery_notification_delay', 'obsess_over_host',
'passive_checks_enabled', 'process_perf_data', 'retain_nonstatus_information',
'retain_status_information', 'retry_check_interval', 'snmp_community',
'snmp_version', 'stalking_options', 'statusmap_image', 'host_notification_options',
'timezone']
return self.getparam(centreon_host_params)
[docs]class HostMacro(hostfactory.ObjHostMacro):
def __init__(self, properties):
super(hostfactory.ObjHostMacro, self).__init__()
self.name = properties.get('macro name')
self.value = properties.get('macro value')
self.description = properties.get('description', '')
self.is_password = properties.get('is_password')
self.source = properties.get('source')
if self.is_password == '':
self.is_password = 0
self.engine_name = '$_HOST' + self.name + '$'
def __repr__(self):
return self.engine_name
def __str__(self):
return self.engine_name
[docs]class Hosts(common.CentreonClass):
"""
Centreon Web host object
"""
def __init__(self):
super(Hosts, self).__init__()
self.hosts = {}
self.__clapi_action = 'HOST'
def __contains__(self, name):
return name in self.hosts.keys() or None
def __getitem__(self, name):
if not self.hosts:
self.list(name)
if name in self.hosts.keys():
return True, self.hosts[name]
else:
return False, None
def _refresh_list(self, name=None):
self.hosts.clear()
state, host = self.webservice.call_clapi(
'show',
self.__clapi_action,
name)
if state and len(host['result']) > 0:
for h in host['result']:
host_obj = Host(h)
self.hosts[host_obj.name] = host_obj
[docs] def list(self, name=None):
self._refresh_list(name)
return self.hosts
[docs] def add(self,
name,
alias,
ip,
instance=None,
template=None,
hg=None):
"""
Add new Host on Centreon platform
:param name: name for host
:param alias: alias (short name for example)
:param ip: Ip address or DNS
:param instance: Poller() or str()
:param template: HostTemplate(), list() of HostTemplate(),
list() of str() or str()
:param hg: HostGroup(), list() of HostGroup(),
list() of str() or str()
:return:
"""
values = [
name,
alias,
ip,
str("|".join(hostfactory.build_param_hosttemplate(template))) if template else template,
str(pollerfactory.build_param_poller(instance)[0]) if instance else "Central",
str("|".join(hostgroupfactory.build_param_hostgroup(hg))) if hg else hg
]
return self.webservice.call_clapi(
'add',
self.__clapi_action,
values)
[docs] def delete(self, host):
value = str(common.build_param(host, Host)[0])
state, delete = self.webservice.call_clapi(
'del',
self.__clapi_action,
value)
if state:
self.hosts.pop(value, None)
return state, delete
[docs]class HostTemplates(Hosts):
def __init__(self):
super(HostTemplates, self).__init__()
self.__clapi_action = 'HTPL'
[docs]class HostTemplate(hostfactory.ObjHostTemplate, Host):
def __init__(self, properties):
super(HostTemplate, self).__init__(properties)
self.__clapi_action = 'HTPL'