1 # -*- coding: utf-8 -*-
2 # Description: netdata python modules framework
3 # Author: Pawel Krupa (paulfantom)
6 # ALL CODE NEEDS TO BE COMPATIBLE WITH Python > 2.7 and Python > 3.1
7 # Follow PEP8 as much as it is possible
8 # "check" and "create" CANNOT be blocking.
9 # "update" CAN be blocking
10 # "update" function needs to be fast, so follow:
11 # https://wiki.python.org/moin/PythonSpeed/PerformanceTips
13 # - use local variables wherever it is possible
14 # - avoid dots in expressions that are executed many times
15 # - use "join()" instead of "+"
16 # - use "import" only at the beginning
18 # using ".encode()" in one thread can block other threads as well (only in python2)
26 import urllib.request as urllib2
30 from subprocess import Popen, PIPE
36 # class BaseService(threading.Thread):
37 class SimpleService(threading.Thread):
39 Prototype of Service class.
40 Implemented basic functionality to run jobs by `python.d.plugin`
42 def __init__(self, configuration=None, name=None):
44 This needs to be initialized in child classes
45 :param configuration: dict
48 threading.Thread.__init__(self)
49 self._data_stream = ""
53 self.priority = 140000
56 self.override_name = None
60 self.__chart_set = False
61 self.__first_run = True
64 if configuration is None:
65 self.error("BaseService: no configuration parameters supplied. Cannot create Service.")
68 self._extract_base_config(configuration)
70 self.create_timetable()
72 # --- BASIC SERVICE CONFIGURATION ---
74 def _extract_base_config(self, config):
76 Get basic parameters to run service
78 config = {'update_every':1,
85 self.override_name = pop('name')
88 self.update_every = int(pop('update_every'))
89 self.priority = int(pop('priority'))
90 self.retries = int(pop('retries'))
91 self.retries_left = self.retries
92 self.configuration = config
94 def create_timetable(self, freq=None):
96 Create service timetable.
99 timetable = {'last': 1466370091.3767564,
105 freq = self.update_every
107 self.timetable = {'last': now,
108 'next': now - (now % freq) + freq,
111 # --- THREAD CONFIGURATION ---
115 Executes self.update(interval) and draws run time chart.
116 Return value presents exit status of update()
119 t_start = time.time()
120 timetable = self.timetable
121 chart_name = self.chart_name
122 # check if it is time to execute job update() function
123 if timetable['next'] > t_start:
124 self.debug(chart_name, "will be run in", str(int((timetable['next'] - t_start) * 1000)), "ms")
127 since_last = int((t_start - timetable['last']) * 1000000)
128 self.debug(chart_name,
129 "ready to run, after", str(int((t_start - timetable['last']) * 1000)),
130 "ms (update_every:", str(timetable['freq'] * 1000),
131 "ms, latency:", str(int((t_start - timetable['next']) * 1000)), "ms")
134 if not self.update(since_last):
135 self.error("update function failed.")
138 self.timetable['next'] = t_end - (t_end % timetable['freq']) + timetable['freq']
139 # draw performance graph
140 run_time = str(int((t_end - t_start) * 1000))
141 # noinspection SqlNoDataSourceInspection
142 print("BEGIN netdata.plugin_pythond_%s %s\nSET run_time = %s\nEND\n" %
143 (self.chart_name, str(since_last), run_time))
144 # sys.stdout.write("BEGIN netdata.plugin_pythond_%s %s\nSET run_time = %s\nEND\n" %
145 # (self.chart_name, str(since_last), run_time))
147 self.debug(chart_name, "updated in", str(run_time), "ms")
148 self.timetable['last'] = t_start
149 self.__first_run = False
154 Runs job in thread. Handles retries.
155 Exits when job failed or timed out.
158 self.timetable['last'] = time.time()
159 while True: # run forever, unless something is wrong
161 status = self._run_once()
162 except Exception as e:
163 self.error("Something wrong: ", str(e))
165 if status: # handle retries if update failed
166 time.sleep(self.timetable['next'] - time.time())
167 self.retries_left = self.retries
169 self.retries_left -= 1
170 if self.retries_left <= 0:
171 self.error("no more retries. Exiting")
174 time.sleep(self.timetable['freq'])
181 Escape and convert passed arguments.
182 :param args: anything
186 append = params.append
191 if type(p) is not str:
198 def _line(self, instruction, *params):
200 Converts *params to string and joins them with one space between every one.
201 Result is appended to self._data_stream
202 :param params: str/int/float
204 tmp = list(map((lambda x: "''" if x is None or len(x) == 0 else x), params))
205 self._data_stream += "%s %s\n" % (instruction, str(" ".join(tmp)))
207 def chart(self, type_id, name="", title="", units="", family="",
208 category="", chart_type="line", priority="", update_every=""):
217 :param chart_type: str
218 :param priority: int/str
219 :param update_every: int/str
221 self._charts.append(type_id)
223 p = self._format(type_id, name, title, units, family, category, chart_type, priority, update_every)
224 self._line("CHART", *p)
226 def dimension(self, id, name=None, algorithm="absolute", multiplier=1, divisor=1, hidden=False):
228 Defines a new dimension for the chart
231 :param algorithm: str
232 :param multiplier: int/str
233 :param divisor: int/str
234 :param hidden: boolean
240 self.error("malformed dimension: multiplier is not a number:", multiplier)
245 self.error("malformed dimension: divisor is not a number:", divisor)
249 if algorithm not in ("absolute", "incremental", "percentage-of-absolute-row", "percentage-of-incremental-row"):
250 algorithm = "absolute"
252 self._dimensions.append(str(id))
254 p = self._format(id, name, algorithm, multiplier, divisor, "hidden")
256 p = self._format(id, name, algorithm, multiplier, divisor)
258 self._line("DIMENSION", *p)
260 def begin(self, type_id, microseconds=0):
264 :param microseconds: int
267 if type_id not in self._charts:
268 self.error("wrong chart type_id:", type_id)
273 self.error("malformed begin statement: microseconds are not a number:", microseconds)
276 self._line("BEGIN", type_id, str(microseconds))
279 def set(self, id, value):
281 Set value to dimension
283 :param value: int/float
286 if id not in self._dimensions:
287 self.error("wrong dimension id:", id, "Available dimensions are:", *self._dimensions)
290 value = str(int(value))
292 self.error("cannot set non-numeric value:", str(value))
294 self._line("SET", id, "=", str(value))
295 self.__chart_set = True
301 self.__chart_set = False
303 pos = self._data_stream.rfind("BEGIN")
304 self._data_stream = self._data_stream[:pos]
308 Upload new data to netdata.
310 print(self._data_stream)
311 self._data_stream = ""
313 # --- ERROR HANDLING ---
315 def error(self, *params):
317 Show error message on stderr
319 msg.error(self.chart_name, *params)
321 def debug(self, *params):
323 Show debug message on stderr
325 msg.debug(self.chart_name, *params)
327 def info(self, *params):
329 Show information message on stderr
331 msg.info(self.chart_name, *params)
333 # --- MAIN METHODS ---
347 self.debug("Module", str(self.__module__), "doesn't implement check() function. Using default.")
348 if self._get_data() is None or len(self._get_data()) == 0:
358 data = self._get_data()
363 for name in self.order:
364 options = self.definitions[name]['options'] + [self.priority + idx, self.update_every]
365 self.chart(self.chart_name + "." + name, *options)
366 # check if server has this datapoint
367 for line in self.definitions[name]['lines']:
369 self.dimension(*line)
375 def update(self, interval):
381 data = self._get_data()
383 self.debug("_get_data() returned no data")
387 for chart in self.order:
388 if self.begin(self.chart_name + "." + chart, interval):
390 for dim in self.definitions[chart]['lines']:
392 self.set(dim[0], data[dim[0]])
399 self.error("no charts to update")
404 class UrlService(SimpleService):
405 # TODO add support for https connections
406 def __init__(self, configuration=None, name=None):
411 SimpleService.__init__(self, configuration=configuration, name=name)
413 def __add_openers(self):
414 # TODO add error handling
415 self.opener = urllib2.build_opener()
418 # TODO currently self.proxies isn't parsed from configuration file
419 # if len(self.proxies) > 0:
420 # for proxy in self.proxies:
423 # if "user" in proxy and "pass" in proxy:
424 # if url.lower().startswith('https://'):
425 # url = 'https://' + proxy['user'] + ':' + proxy['pass'] + '@' + url[8:]
427 # url = 'http://' + proxy['user'] + ':' + proxy['pass'] + '@' + url[7:]
428 # # FIXME move proxy auth to sth like this:
429 # # passman = urllib2.HTTPPasswordMgrWithDefaultRealm()
430 # # passman.add_password(None, url, proxy['user'], proxy['password'])
431 # # opener.add_handler(urllib2.HTTPBasicAuthHandler(passman))
433 # if url.lower().startswith('https://'):
434 # opener.add_handler(urllib2.ProxyHandler({'https': url}))
436 # opener.add_handler(urllib2.ProxyHandler({'https': url}))
439 if self.user is not None and self.password is not None:
440 passman = urllib2.HTTPPasswordMgrWithDefaultRealm()
441 passman.add_password(None, self.url, self.user, self.password)
442 self.opener.add_handler(urllib2.HTTPBasicAuthHandler(passman))
443 self.debug("Enabling HTTP basic auth")
445 #urllib2.install_opener(opener)
447 def _get_raw_data(self):
449 Get raw data from http request
454 f = self.opener.open(self.url, timeout=self.update_every * 2)
455 # f = urllib2.urlopen(self.url, timeout=self.update_every * 2)
456 except Exception as e:
461 raw = f.read().decode('utf-8')
462 except Exception as e:
470 Format configuration data and try to connect to server
473 if self.name is None or self.name == str(None):
475 self.chart_name += "_" + self.name
477 self.name = str(self.name)
479 self.url = str(self.configuration['url'])
480 except (KeyError, TypeError):
483 self.user = str(self.configuration['user'])
484 except (KeyError, TypeError):
487 self.password = str(self.configuration['pass'])
488 except (KeyError, TypeError):
493 test = self._get_data()
494 if test is None or len(test) == 0:
500 class SocketService(SimpleService):
501 def __init__(self, configuration=None, name=None):
503 self._keep_alive = False
504 self.host = "localhost"
506 self.unix_socket = None
508 self.__socket_config = None
509 SimpleService.__init__(self, configuration=configuration, name=name)
513 Recreate socket and connect to it since sockets cannot be reused after closing
514 Available configurations are IPv6, IPv4 or UNIX socket
518 if self.unix_socket is None:
519 if self.__socket_config is None:
520 # establish ipv6 or ipv4 connection.
521 for res in socket.getaddrinfo(self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM):
523 # noinspection SpellCheckingInspection
524 af, socktype, proto, canonname, sa = res
525 self._sock = socket.socket(af, socktype, proto)
526 except socket.error as e:
527 self.debug("Cannot create socket:", str(e))
531 self._sock.connect(sa)
532 except socket.error as e:
533 self.debug("Cannot connect to socket:", str(e))
536 self.__socket_config = res
539 # connect to socket with previously established configuration
541 af, socktype, proto, canonname, sa = self.__socket_config
542 self._sock = socket.socket(af, socktype, proto)
543 self._sock.connect(sa)
544 except socket.error as e:
545 self.debug("Cannot create or connect to socket:", str(e))
548 # connect to unix socket
550 self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
551 self._sock.connect(self.unix_socket)
553 self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
554 self._sock.connect(self.unix_socket)
556 except Exception as e:
558 "Cannot create socket with following configuration: host:", str(self.host),
559 "port:", str(self.port),
560 "socket:", str(self.unix_socket))
562 self._sock.setblocking(0)
564 def _disconnect(self):
566 Close socket connection
570 self._sock.shutdown(2) # 0 - read, 1 - write, 2 - all
581 # Send request if it is needed
582 if self.request != "".encode():
584 self._sock.send(self.request)
585 except Exception as e:
588 "used configuration: host:", str(self.host),
589 "port:", str(self.port),
590 "socket:", str(self.unix_socket))
596 Receive data from socket
602 ready_to_read, _, in_error = select.select([self._sock], [], [], 5)
603 except Exception as e:
604 self.debug("SELECT", str(e))
607 if len(ready_to_read) > 0:
608 buf = self._sock.recv(4096)
609 if len(buf) == 0 or buf is None: # handle server disconnect
611 data += buf.decode(errors='ignore')
612 if self._check_raw_data(data):
615 self.error("Socket timed out.")
621 def _get_raw_data(self):
623 Get raw data with low-level "socket" module.
626 if self._sock is None:
629 # Send request if it is needed
633 data = self._receive()
635 if not self._keep_alive:
640 def _check_raw_data(self, data):
642 Check if all data has been gathered from socket
648 def _parse_config(self):
650 Parse configuration data
653 if self.name is None or self.name == str(None):
656 self.name = str(self.name)
658 self.unix_socket = str(self.configuration['socket'])
659 except (KeyError, TypeError):
660 self.debug("No unix socket specified. Trying TCP/IP socket.")
662 self.host = str(self.configuration['host'])
663 except (KeyError, TypeError):
664 self.debug("No host specified. Using: '" + self.host + "'")
666 self.port = int(self.configuration['port'])
667 except (KeyError, TypeError):
668 self.debug("No port specified. Using: '" + str(self.port) + "'")
670 self.request = str(self.configuration['request'])
671 except (KeyError, TypeError):
672 self.debug("No request specified. Using: '" + str(self.request) + "'")
673 self.request = self.request.encode()
676 return SimpleService.check(self)
679 class LogService(SimpleService):
680 def __init__(self, configuration=None, name=None):
682 self._last_position = 0
683 # self._log_reader = None
684 SimpleService.__init__(self, configuration=configuration, name=name)
685 self.retries = 100000 # basically always retry
687 def _get_raw_data(self):
689 Get log lines since last poll
694 if os.path.getsize(self.log_path) < self._last_position:
695 self._last_position = 0 # read from beginning if file has shrunk
696 elif os.path.getsize(self.log_path) == self._last_position:
697 self.debug("Log file hasn't changed. No new data.")
698 return [] # return empty list if nothing has changed
699 with open(self.log_path, "r") as fp:
700 fp.seek(self._last_position)
701 for i, line in enumerate(fp):
703 self._last_position = fp.tell()
704 except Exception as e:
710 self.error("No data collected.")
715 Parse basic configuration and check if log file exists
718 if self.name is not None or self.name != str(None):
721 self.name = str(self.name)
723 self.log_path = str(self.configuration['path'])
724 except (KeyError, TypeError):
725 self.error("No path to log specified. Using: '" + self.log_path + "'")
727 if os.access(self.log_path, os.R_OK):
730 self.error("Cannot access file: '" + self.log_path + "'")
734 # set cursor at last byte of log file
735 self._last_position = os.path.getsize(self.log_path)
736 status = SimpleService.create(self)
737 # self._last_position = 0
741 class ExecutableService(SimpleService):
742 bad_substrings = ('&', '|', ';', '>', '<')
744 def __init__(self, configuration=None, name=None):
746 SimpleService.__init__(self, configuration=configuration, name=name)
748 def _get_raw_data(self):
750 Get raw data from executed command
754 p = Popen(self.command, stdout=PIPE, stderr=PIPE)
755 except Exception as e:
756 self.error("Executing command", self.command, "resulted in error:", str(e))
759 for line in p.stdout.readlines():
760 data.append(str(line.decode()))
763 self.error("No data collected.")
770 Parse basic configuration, check if command is whitelisted and is returning values
773 if self.name is not None or self.name != str(None):
776 self.name = str(self.name)
778 self.command = str(self.configuration['command'])
779 except (KeyError, TypeError):
780 self.error("No command specified. Using: '" + self.command + "'")
781 command = self.command.split(' ')
783 for arg in command[1:]:
784 if any(st in arg for st in self.bad_substrings):
785 self.error("Bad command argument:" + " ".join(self.command[1:]))
787 # test command and search for it in /usr/sbin or /sbin when failed
788 base = command[0].split('/')[-1]
789 if self._get_raw_data() is None:
790 for prefix in ['/sbin/', '/usr/sbin/']:
791 command[0] = prefix + base
792 if os.path.isfile(command[0]):
794 self.command = command
795 if self._get_data() is None or len(self._get_data()) == 0:
796 self.error("Command", self.command, "returned no data")