Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def getMetricData(config, parameters, hosts, startTime, endTime):
"""Get metric data from Zabbix API"""
# connection to zabbix
logger.info(
"Begin to connect to zabbix:{} User/Pwd:{}/{}".format(str(config['ZABBIX_URL']), str(config['ZABBIX_USER']),
str(config['ZABBIX_PASSWORD'])))
zapi = ZabbixAPI(url=config['ZABBIX_URL'], user=config['ZABBIX_USER'], password=config['ZABBIX_PASSWORD'])
logger.info("Get connection from zabbix success")
# get hosts
hosts_map = {}
hosts_ids = []
hosts_res = zapi.do_request('host.get', {
'filter': {
"host": hosts,
},
"sortfield": "name",
'output': 'extend'
})
for item in hosts_res['result']:
host_id = item['hostid']
host = item['host']
hosts_ids.append(host_id)
def zabbix_network_get():
db_zabbix = db_idc.zabbix_info
zapi = ZabbixAPI(url=zabbix_url, user=zabbix_user, password=zabbix_pw)
dt = datetime.datetime.now()
now_time = time.mktime(dt.timetuple())
old_time = dt - datetime.timedelta(days=1)
old_time = time.mktime(old_time.timetuple())
try:
result = zapi.host.get(monitored_hosts=1, output='extend')
results = {infos['hostid']:infos['host'] for infos in result}
hostids = [infos['hostid'] for infos in result]
items = zapi.item.get(hostids=hostids, output=["hostid", "itemid"], search={"key_": 'net.if.'})
for item in items:
key = "op_zabbix_network_hostid_%s" %item["hostid"]
RC_CLUSTER.lpush(key,item["itemid"])
itemids = [info['itemid'] for info in items]
vals = zapi.trend.get(itemids=itemids, time_from=old_time, time_till=now_time, output=["itemid", "value_max"])
for val in vals:
key = "op_zabbix_network_itemid_%s" %val["itemid"]
from alerts import Alerter # , BasicMatchString
import logging
from pyzabbix.api import ZabbixAPI
from pyzabbix import ZabbixSender, ZabbixMetric
from datetime import datetime
class ZabbixClient(ZabbixAPI):
def __init__(self, url='http://localhost', use_authenticate=False, user='Admin', password='zabbix', sender_host='localhost',
sender_port=10051):
self.url = url
self.use_authenticate = use_authenticate
self.sender_host = sender_host
self.sender_port = sender_port
self.metrics_chunk_size = 200
self.aggregated_metrics = []
self.logger = logging.getLogger(self.__class__.__name__)
super(ZabbixClient, self).__init__(url=self.url, use_authenticate=self.use_authenticate, user=user, password=password)
def send_metric(self, hostname, key, data):
zm = ZabbixMetric(hostname, key, data)
if self.send_aggregated_metrics: