Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from . import models, utils
logger = logging.getLogger(__name__)
sms_settings = getattr(django_settings, 'WALDUR_ZABBIX', {}).get('SMS_SETTINGS', {})
class ZabbixLogsFilter(logging.Filter):
def filter(self, record):
# Mute useless Zabbix log concerning JSON-RPC server endpoint.
if record.getMessage().startswith('JSON-RPC Server Endpoint'):
return False
return super(ZabbixLogsFilter, self).filter(record)
pyzabbix.logger.addFilter(ZabbixLogsFilter())
class ZabbixBackendError(ServiceBackendError):
pass
class ZabbixBackend(ServiceBackend):
DEFAULTS = {
'host_group_name': 'waldur',
'templates_names': [],
'database_parameters': {
'host': 'localhost',
'port': '3306',
'name': 'zabbix',
'user': 'admin',
import warnings
import requests
from requests.exceptions import RequestException
from requests.packages.urllib3 import exceptions
from django.conf import settings as django_settings
from django.utils import six
import pyzabbix
from nodeconductor.monitoring.zabbix.errors import ZabbixError
from nodeconductor.monitoring.zabbix.log import ZabbixLogsFilter
logger = logging.getLogger(__name__)
pyzabbix.logger.addFilter(ZabbixLogsFilter())
def _exception_decorator(message, fail_silently=None):
def decorator(func):
def wrapper(self, *args, **kwargs):
try:
return func(self, *args, **kwargs)
except (pyzabbix.ZabbixAPIException, RequestException) as exception:
if not self._settings.get('FAIL_SILENTLY', False):
exception_name = exception.__class__.__name__
message_args = (self,) + args + tuple(kwargs.values())
logger.exception(message.format(*message_args, exception=exception, exception_name=exception_name))
six.reraise(ZabbixError, exception)
return wrapper