# -*- coding: utf-8 -*-
# Description: netdata python modules framework
# Author: Pawel Krupa (paulfantom)
# Remember:
# ALL CODE NEEDS TO BE COMPATIBLE WITH Python > 2.7 and Python > 3.1
# Follow PEP8 as much as it is possible
# "check" and "create" CANNOT be blocking.
# "update" CAN be blocking
# "update" function needs to be fast, so follow:
# https://wiki.python.org/moin/PythonSpeed/PerformanceTips
# basically:
# - use local variables wherever it is possible
# - avoid dots in expressions that are executed many times
# - use "join()" instead of "+"
# - use "import" only at the beginning
#
# using ".encode()" in one thread can block other threads as well (only in python2)
import os
import re
import socket
import time
import threading
import urllib3
from glob import glob
from subprocess import Popen, PIPE
from sys import exc_info
try:
import MySQLdb
PY_MYSQL = True
except ImportError:
try:
import pymysql as MySQLdb
PY_MYSQL = True
except ImportError:
PY_MYSQL = False
import msg
PATH = os.getenv('PATH', '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin').split(':')
try:
urllib3.disable_warnings()
except AttributeError:
msg.error('urllib3: warnings were not disabled')
# class BaseService(threading.Thread):
class SimpleService(threading.Thread):
"""
Prototype of Service class.
Implemented basic functionality to run jobs by `python.d.plugin`
"""
def __init__(self, configuration=None, name=None):
"""
This needs to be initialized in child classes
:param configuration: dict
:param name: str
"""
threading.Thread.__init__(self)
self._data_stream = ""
self.daemon = True
self.retries = 0
self.retries_left = 0
self.priority = 140000
self.update_every = 1
self.name = name
self.override_name = None
self.chart_name = ""
self._dimensions = []
self._charts = []
self.__chart_set = False
self.__first_run = True
self.order = []
self.definitions = {}
self._data_from_check = dict()
if configuration is None:
self.error("BaseService: no configuration parameters supplied. Cannot create Service.")
raise RuntimeError
else:
self._extract_base_config(configuration)
self.timetable = {}
self.create_timetable()
# --- BASIC SERVICE CONFIGURATION ---
def _extract_base_config(self, config):
"""
Get basic parameters to run service
Minimum config:
config = {'update_every':1,
'priority':100000,
'retries':0}
:param config: dict
"""
pop = config.pop
try:
self.override_name = pop('name')
except KeyError:
pass
self.update_every = int(pop('update_every'))
self.priority = int(pop('priority'))
self.retries = int(pop('retries'))
self.retries_left = self.retries
self.configuration = config
def create_timetable(self, freq=None):
"""
Create service timetable.
`freq` is optional
Example:
timetable = {'last': 1466370091.3767564,
'next': 1466370092,
'freq': 1}
:param freq: int
"""
if freq is None:
freq = self.update_every
now = time.time()
self.timetable = {'last': now,
'next': now - (now % freq) + freq,
'freq': freq}
# --- THREAD CONFIGURATION ---
def _run_once(self):
"""
Executes self.update(interval) and draws run time chart.
Return value presents exit status of update()
:return: boolean
"""
t_start = float(time.time())
chart_name = self.chart_name
since_last = int((t_start - self.timetable['last']) * 1000000)
if self.__first_run:
since_last = 0
if not self.update(since_last):
self.error("update function failed.")
return False
# draw performance graph
run_time = int((time.time() - t_start) * 1000)
print("BEGIN netdata.plugin_pythond_%s %s\nSET run_time = %s\nEND\n" %
(self.chart_name, str(since_last), str(run_time)))
self.debug(chart_name, "updated in", str(run_time), "ms")
self.timetable['last'] = t_start
self.__first_run = False
return True
def run(self):
"""
Runs job in thread. Handles retries.
Exits when job failed or timed out.
:return: None
"""
step = float(self.timetable['freq'])
penalty = 0
self.timetable['last'] = float(time.time() - step)
self.debug("starting data collection - update frequency:", str(step), " retries allowed:", str(self.retries))
while True: # run forever, unless something is wrong
now = float(time.time())
next = self.timetable['next'] = now - (now % step) + step + penalty
# it is important to do this in a loop
# sleep() is interruptable
while now < next:
self.debug("sleeping for", str(next - now), "secs to reach frequency of",
str(step), "secs, now:", str(now), " next:", str(next), " penalty:", str(penalty))
time.sleep(next - now)
now = float(time.time())
# do the job
try:
status = self._run_once()
except Exception:
status = False
if status:
# it is good
self.retries_left = self.retries
penalty = 0
else:
# it failed
self.retries_left -= 1
if self.retries_left 600:
penalty = 600
self.retries_left = self.retries
self.alert("failed to collect data for " + str(self.retries) +
" times - increasing penalty to " + str(penalty) + " sec and trying again")
else:
self.error("failed to collect data - " + str(self.retries_left)
+ " retries left - penalty: " + str(penalty) + " sec")
# --- CHART ---
@staticmethod
def _format(*args):
"""
Escape and convert passed arguments.
:param args: anything
:return: list
"""
params = []
append = params.append
for p in args:
if p is None:
append(p)
continue
if type(p) is not str:
p = str(p)
if ' ' in p:
p = "'" + p + "'"
append(p)
return params
def _line(self, instruction, *params):
"""
Converts *params to string and joins them with one space between every one.
Result is appended to self._data_stream
:param params: str/int/float
"""
tmp = list(map((lambda x: "''" if x is None or len(x) == 0 else x), params))
self._data_stream += "%s %s\n" % (instruction, str(" ".join(tmp)))
def chart(self, type_id, name="", title="", units="", family="",
category="", chart_type="line", priority="", update_every=""):
"""
Defines a new chart.
:param type_id: str
:param name: str
:param title: str
:param units: str
:param family: str
:param category: str
:param chart_type: str
:param priority: int/str
:param update_every: int/str
"""
self._charts.append(type_id)
p = self._format(type_id, name, title, units, family, category, chart_type, priority, update_every)
self._line("CHART", *p)
def dimension(self, id, name=None, algorithm="absolute", multiplier=1, divisor=1, hidden=False):
"""
Defines a new dimension for the chart
:param id: str
:param name: str
:param algorithm: str
:param multiplier: int/str
:param divisor: int/str
:param hidden: boolean
:return:
"""
try:
int(multiplier)
except TypeError:
self.error("malformed dimension: multiplier is not a number:", multiplier)
multiplier = 1
try:
int(divisor)
except TypeError:
self.error("malformed dimension: divisor is not a number:", divisor)
divisor = 1
if name is None:
name = id
if algorithm not in ("absolute", "incremental", "percentage-of-absolute-row", "percentage-of-incremental-row"):
algorithm = "absolute"
self._dimensions.append(str(id))
if hidden:
p = self._format(id, name, algorithm, multiplier, divisor, "hidden")
else:
p = self._format(id, name, algorithm, multiplier, divisor)
self._line("DIMENSION", *p)
def begin(self, type_id, microseconds=0):
"""
Begin data set
:param type_id: str
:param microseconds: int
:return: boolean
"""
if type_id not in self._charts:
self.error("wrong chart type_id:", type_id)
return False
try:
int(microseconds)
except TypeError:
self.error("malformed begin statement: microseconds are not a number:", microseconds)
microseconds = ""
self._line("BEGIN", type_id, str(microseconds))
return True
def set(self, id, value):
"""
Set value to dimension
:param id: str
:param value: int/float
:return: boolean
"""
if id not in self._dimensions:
self.error("wrong dimension id:", id, "Available dimensions are:", *self._dimensions)
return False
try:
value = str(int(value))
except TypeError:
self.error("cannot set non-numeric value:", str(value))
return False
self._line("SET", id, "=", str(value))
self.__chart_set = True
return True
def end(self):
if self.__chart_set:
self._line("END")
self.__chart_set = False
else:
pos = self._data_stream.rfind("BEGIN")
self._data_stream = self._data_stream[:pos]
def commit(self):
"""
Upload new data to netdata.
"""
try:
print(self._data_stream)
except Exception as e:
msg.fatal('cannot send data to netdata:', str(e))
self._data_stream = ""
# --- ERROR HANDLING ---
def error(self, *params):
"""
Show error message on stderr
"""
msg.error(self.chart_name, *params)
def alert(self, *params):
"""
Show error message on stderr
"""
msg.alert(self.chart_name, *params)
def debug(self, *params):
"""
Show debug message on stderr
"""
msg.debug(self.chart_name, *params)
def info(self, *params):
"""
Show information message on stderr
"""
msg.info(self.chart_name, *params)
# --- MAIN METHODS ---
def _get_data(self):
"""
Get some data
:return: dict
"""
return {}
def check(self):
"""
check() prototype
:return: boolean
"""
self.debug("Module", str(self.__module__), "doesn't implement check() function. Using default.")
data = self._get_data()
if data is None:
self.debug("failed to receive data during check().")
return False
if len(data) == 0:
self.debug("empty data during check().")
return False
self.debug("successfully received data during check(): '" + str(data) + "'")
return True
def create(self):
"""
Create charts
:return: boolean
"""
data = self._data_from_check or self._get_data()
if data is None:
self.debug("failed to receive data during create().")
return False
idx = 0
for name in self.order:
options = self.definitions[name]['options'] + [self.priority + idx, self.update_every]
self.chart(self.chart_name + "." + name, *options)
# check if server has this datapoint
for line in self.definitions[name]['lines']:
if line[0] in data:
self.dimension(*line)
idx += 1
self.commit()
return True
def update(self, interval):
"""
Update charts
:param interval: int
:return: boolean
"""
data = self._get_data()
if data is None:
self.debug("failed to receive data during update().")
return False
updated = False
for chart in self.order:
if self.begin(self.chart_name + "." + chart, interval):
updated = True
for dim in self.definitions[chart]['lines']:
try:
self.set(dim[0], data[dim[0]])
except KeyError:
pass
self.end()
self.commit()
if not updated:
self.error("no charts to update")
return updated
@staticmethod
def find_binary(binary):
try:
if isinstance(binary, str):
binary = os.path.basename(binary)
return next(('/'.join([p, binary]) for p in PATH
if os.path.isfile('/'.join([p, binary]))
and os.access('/'.join([p, binary]), os.X_OK)))
return None
except StopIteration:
return None
def _add_new_dimension(self, dimension_id, chart_name, dimension=None, algorithm='incremental',
multiplier=1, divisor=1, priority=65000):
"""
:param dimension_id:
:param chart_name:
:param dimension:
:param algorithm:
:param multiplier:
:param divisor:
:param priority:
:return:
"""
if not all([dimension_id not in self._dimensions,
chart_name in self.order,
chart_name in self.definitions]):
return
self._dimensions.append(dimension_id)
dimension_list = list(map(str, [dimension_id,
dimension if dimension else dimension_id,
algorithm,
multiplier,
divisor]))
self.definitions[chart_name]['lines'].append(dimension_list)
add_to_name = self.override_name or self.name
job_name = ('_'.join([self.__module__, re.sub('\s+', '_', add_to_name)])
if add_to_name != 'None' else self.__module__)
chart = 'CHART {0}.{1} '.format(job_name, chart_name)
options = '"" "{0}" {1} "{2}" {3} {4} '.format(*self.definitions[chart_name]['options'][1:6])
other = '{0} {1}\n'.format(priority, self.update_every)
new_dimension = "DIMENSION {0}\n".format(' '.join(dimension_list))
print(chart + options + other + new_dimension)
class UrlService(SimpleService):
def __init__(self, configuration=None, name=None):
SimpleService.__init__(self, configuration=configuration, name=name)
self.url = self.configuration.get('url')
self.user = self.configuration.get('user')
self.password = self.configuration.get('pass')
self.proxy_user = self.configuration.get('proxy_user')
self.proxy_password = self.configuration.get('proxy_pass')
self.proxy_url = self.configuration.get('proxy_url')
self._manager = None
def __make_headers(self, **header_kw):
user = header_kw.get('user') or self.user
password = header_kw.get('pass') or self.password
proxy_user = header_kw.get('proxy_user') or self.proxy_user
proxy_password = header_kw.get('proxy_pass') or self.proxy_password
header_params = dict(keep_alive=True)
proxy_header_params = dict()
if user and password:
header_params['basic_auth'] = '{user}:{password}'.format(user=user,
password=password)
if proxy_user and proxy_password:
proxy_header_params['proxy_basic_auth'] = '{user}:{password}'.format(user=proxy_user,
password=proxy_password)
try:
return urllib3.make_headers(**header_params), urllib3.make_headers(**proxy_header_params)
except TypeError as error:
self.error('build_header() error: {error}'.format(error=error))
return None, None
def _build_manager(self, **header_kw):
header, proxy_header = self.__make_headers(**header_kw)
if header is None or proxy_header is None:
return None
proxy_url = header_kw.get('proxy_url') or self.proxy_url
if proxy_url:
manager = urllib3.ProxyManager
params = dict(proxy_url=proxy_url, headers=header, proxy_headers=proxy_header)
else:
manager = urllib3.PoolManager
params = dict(headers=header)
try:
return manager(**params)
except (urllib3.exceptions.ProxySchemeUnknown, TypeError) as error:
self.error('build_manager() error:', str(error))
return None
def _get_raw_data(self, url=None, manager=None):
"""
Get raw data from http request
:return: str
"""
try:
url = url or self.url
manager = manager or self._manager
# TODO: timeout, retries and method hardcoded..
response = manager.request(method='GET',
url=url,
timeout=1,
retries=1,
headers=manager.headers)
except (urllib3.exceptions.HTTPError, TypeError, AttributeError) as error:
self.error('Url: {url}. Error: {error}'.format(url=url, error=error))
return None
if response.status == 200:
return response.data.decode()
self.debug('Url: {url}. Http response status code: {code}'.format(url=url, code=response.status))
return None
def check(self):
"""
Format configuration data and try to connect to server
:return: boolean
"""
if not (self.url and isinstance(self.url, str)):
self.error('URL is not defined or type is not ')
return False
self._manager = self._build_manager()
if not self._manager:
return False
try:
data = self._get_data()
except Exception as error:
self.error('_get_data() failed. Url: {url}. Error: {error}'.format(url=self.url, error=error))
return False
if isinstance(data, dict) and data:
self._data_from_check = data
return True
self.error('_get_data() returned no data or type is not ')
return False
class SocketService(SimpleService):
def __init__(self, configuration=None, name=None):
self._sock = None
self._keep_alive = False
self.host = "localhost"
self.port = None
self.unix_socket = None
self.request = ""
self.__socket_config = None
self.__empty_request = "".encode()
SimpleService.__init__(self, configuration=configuration, name=name)
def _socketerror(self, message=None):
if self.unix_socket is not None:
self.error("unix socket '" + self.unix_socket + "':", message)
else:
if self.__socket_config is not None:
af, socktype, proto, canonname, sa = self.__socket_config
self.error("socket to '" + str(sa[0]) + "' port " + str(sa[1]) + ":", message)
else:
self.error("unknown socket:", message)
def _connect2socket(self, res=None):
"""
Connect to a socket, passing the result of getaddrinfo()
:return: boolean
"""
if res is None:
res = self.__socket_config
if res is None:
self.error("Cannot create socket to 'None':")
return False
af, socktype, proto, canonname, sa = res
try:
self.debug("creating socket to '" + str(sa[0]) + "', port " + str(sa[1]))
self._sock = socket.socket(af, socktype, proto)
except socket.error as e:
self.error("Failed to create socket to '" + str(sa[0]) + "', port " + str(sa[1]) + ":", str(e))
self._sock = None
self.__socket_config = None
return False
try:
self.debug("connecting socket to '" + str(sa[0]) + "', port " + str(sa[1]))
self._sock.connect(sa)
except socket.error as e:
self.error("Failed to connect to '" + str(sa[0]) + "', port " + str(sa[1]) + ":", str(e))
self._disconnect()
self.__socket_config = None
return False
self.debug("connected to '" + str(sa[0]) + "', port " + str(sa[1]))
self.__socket_config = res
return True
def _connect2unixsocket(self):
"""
Connect to a unix socket, given its filename
:return: boolean
"""
if self.unix_socket is None:
self.error("cannot connect to unix socket 'None'")
return False
try:
self.debug("attempting DGRAM unix socket '" + str(self.unix_socket) + "'")
self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
self._sock.connect(self.unix_socket)
self.debug("connected DGRAM unix socket '" + str(self.unix_socket) + "'")
return True
except socket.error as e:
self.debug("Failed to connect DGRAM unix socket '" + str(self.unix_socket) + "':", str(e))
try:
self.debug("attempting STREAM unix socket '" + str(self.unix_socket) + "'")
self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self._sock.connect(self.unix_socket)
self.debug("connected STREAM unix socket '" + str(self.unix_socket) + "'")
return True
except socket.error as e:
self.debug("Failed to connect STREAM unix socket '" + str(self.unix_socket) + "':", str(e))
self.error("Failed to connect to unix socket '" + str(self.unix_socket) + "':", str(e))
self._sock = None
return False
def _connect(self):
"""
Recreate socket and connect to it since sockets cannot be reused after closing
Available configurations are IPv6, IPv4 or UNIX socket
:return:
"""
try:
if self.unix_socket is not None:
self._connect2unixsocket()
else:
if self.__socket_config is not None:
self._connect2socket()
else:
for res in socket.getaddrinfo(self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM):
if self._connect2socket(res): break
except Exception as e:
self._sock = None
self.__socket_config = None
if self._sock is not None:
self._sock.setblocking(0)
self._sock.settimeout(5)
self.debug("set socket timeout to: " + str(self._sock.gettimeout()))
def _disconnect(self):
"""
Close socket connection
:return:
"""
if self._sock is not None:
try:
self.debug("closing socket")
self._sock.shutdown(2) # 0 - read, 1 - write, 2 - all
self._sock.close()
except Exception:
pass
self._sock = None
def _send(self):
"""
Send request.
:return: boolean
"""
# Send request if it is needed
if self.request != self.__empty_request:
try:
self.debug("sending request:", str(self.request))
self._sock.send(self.request)
except Exception as e:
self._socketerror("error sending request:" + str(e))
self._disconnect()
return False
return True
def _receive(self):
"""
Receive data from socket
:return: str
"""
data = ""
while True:
self.debug("receiving response")
try:
buf = self._sock.recv(4096)
except Exception as e:
self._socketerror("failed to receive response:" + str(e))
self._disconnect()
break
if buf is None or len(buf) == 0: # handle server disconnect
if data == "":
self._socketerror("unexpectedly disconnected")
else:
self.debug("server closed the connection")
self._disconnect()
break
self.debug("received data:", str(buf))
data += buf.decode('utf-8', 'ignore')
if self._check_raw_data(data):
break
self.debug("final response:", str(data))
return data
def _get_raw_data(self):
"""
Get raw data with low-level "socket" module.
:return: str
"""
if self._sock is None:
self._connect()
if self._sock is None:
return None
# Send request if it is needed
if not self._send():
return None
data = self._receive()
if not self._keep_alive:
self._disconnect()
return data
def _check_raw_data(self, data):
"""
Check if all data has been gathered from socket
:param data: str
:return: boolean
"""
return True
def _parse_config(self):
"""
Parse configuration data
:return: boolean
"""
if self.name is None or self.name == str(None):
self.name = ""
else:
self.name = str(self.name)
try:
self.unix_socket = str(self.configuration['socket'])
except (KeyError, TypeError):
self.debug("No unix socket specified. Trying TCP/IP socket.")
self.unix_socket = None
try:
self.host = str(self.configuration['host'])
except (KeyError, TypeError):
self.debug("No host specified. Using: '" + self.host + "'")
try:
self.port = int(self.configuration['port'])
except (KeyError, TypeError):
self.debug("No port specified. Using: '" + str(self.port) + "'")
try:
self.request = str(self.configuration['request'])
except (KeyError, TypeError):
self.debug("No request specified. Using: '" + str(self.request) + "'")
self.request = self.request.encode()
def check(self):
self._parse_config()
return SimpleService.check(self)
class LogService(SimpleService):
def __init__(self, configuration=None, name=None):
SimpleService.__init__(self, configuration=configuration, name=name)
self.log_path = self.configuration.get('path')
self.__glob_path = self.log_path
self._last_position = 0
self.retries = 100000 # basically always retry
self.__re_find = dict(current=0, run=0, maximum=60)
def _get_raw_data(self):
"""
Get log lines since last poll
:return: list
"""
lines = list()
try:
if self.__re_find['current'] == self.__re_find['run']:
self._find_recent_log_file()
size = os.path.getsize(self.log_path)
if size == self._last_position:
self.__re_find['current'] += 1
return list() # return empty list if nothing has changed
elif size < self._last_position:
self._last_position = 0 # read from beginning if file has shrunk
with open(self.log_path) as fp:
fp.seek(self._last_position)
for line in fp:
lines.append(line)
self._last_position = fp.tell()
self.__re_find['current'] = 0
except (OSError, IOError) as error:
self.__re_find['current'] += 1
self.error(str(error))
return lines or None
def _find_recent_log_file(self):
"""
:return:
"""
self.__re_find['run'] = self.__re_find['maximum']
self.__re_find['current'] = 0
self.__glob_path = self.__glob_path or self.log_path # workaround for modules w/o config files
path_list = glob(self.__glob_path)
if path_list:
self.log_path = max(path_list)
return True
return False
def check(self):
"""
Parse basic configuration and check if log file exists
:return: boolean
"""
if not self.log_path:
self.error("No path to log specified")
return None
if all([self._find_recent_log_file(),
os.access(self.log_path, os.R_OK),
os.path.isfile(self.log_path)]):
return True
self.error("Cannot access %s" % self.log_path)
return False
def create(self):
# set cursor at last byte of log file
self._last_position = os.path.getsize(self.log_path)
status = SimpleService.create(self)
# self._last_position = 0
return status
class ExecutableService(SimpleService):
def __init__(self, configuration=None, name=None):
SimpleService.__init__(self, configuration=configuration, name=name)
self.command = None
def _get_raw_data(self, stderr=False):
"""
Get raw data from executed command
:return:
"""
try:
p = Popen(self.command, stdout=PIPE, stderr=PIPE)
except Exception as error:
self.error("Executing command", " ".join(self.command), "resulted in error:", str(error))
return None
data = list()
std = p.stderr if stderr else p.stdout
for line in std.readlines():
data.append(line.decode())
return data or None
def check(self):
"""
Parse basic configuration, check if command is whitelisted and is returning values
:return:
"""
# Preference: 1. "command" from configuration file 2. "command" from plugin (if specified)
if 'command' in self.configuration:
self.command = self.configuration['command']
# "command" must be: 1.not None 2. type
if not (self.command and isinstance(self.command, str)):
self.error('Command is not defined or command type is not ')
return False
# Split "command" into: 1. command 2. options
command, opts = self.command.split()[0], self.command.split()[1:]
# Check for "bad" symbols in options. No pipes, redirects etc. TODO: what is missing?
bad_opts = set(''.join(opts)) & set(['&', '|', ';', '>', '