1 # -*- coding: utf-8 -*-
2 # Description: prototypes for netdata python.d modules
3 # Author: Pawel Krupa (paulfantom)
11 import urllib.request as urllib2
15 from subprocess import Popen, PIPE
21 class BaseService(threading.Thread):
23 Prototype of Service class.
24 Implemented basic functionality to run jobs by `python.d.plugin`
26 def __init__(self, configuration=None, name=None):
28 This needs to be initialized in child classes
29 :param configuration: dict
32 threading.Thread.__init__(self)
33 self._data_stream = ""
37 self.priority = 140000
40 self.override_name = None
44 if configuration is None:
45 self.error("BaseService: no configuration parameters supplied. Cannot create Service.")
48 self._extract_base_config(configuration)
50 self.create_timetable()
52 def _extract_base_config(self, config):
54 Get basic parameters to run service
56 config = {'update_every':1,
62 self.override_name = config.pop('name')
65 self.update_every = int(config.pop('update_every'))
66 self.priority = int(config.pop('priority'))
67 self.retries = int(config.pop('retries'))
68 self.retries_left = self.retries
69 self.configuration = config
71 def create_timetable(self, freq=None):
73 Create service timetable.
76 timetable = {'last': 1466370091.3767564,
82 freq = self.update_every
84 self.timetable = {'last': now,
85 'next': now - (now % freq) + freq,
90 Executes self.update(interval) and draws run time chart.
91 Return value presents exit status of update()
95 # check if it is time to execute job update() function
96 if self.timetable['next'] > t_start:
97 #msg.debug(self.chart_name + " will be run in " +
98 # str(int((self.timetable['next'] - t_start) * 1000)) + " ms")
99 msg.debug(self.chart_name, "will be run in", str(int((self.timetable['next'] - t_start) * 1000)), "ms")
102 since_last = int((t_start - self.timetable['last']) * 1000000)
103 #msg.debug(self.chart_name +
104 # " ready to run, after " + str(int((t_start - self.timetable['last']) * 1000)) +
105 # " ms (update_every: " + str(self.timetable['freq'] * 1000) +
106 # " ms, latency: " + str(int((t_start - self.timetable['next']) * 1000)) + " ms)")
107 msg.debug(self.chart_name,
108 "ready to run, after", str(int((t_start - self.timetable['last']) * 1000)),
109 "ms (update_every:", str(self.timetable['freq'] * 1000),
110 "ms, latency:", str(int((t_start - self.timetable['next']) * 1000)), "ms")
111 if not self.update(since_last):
112 self.error("update function failed.")
115 self.timetable['next'] = t_end - (t_end % self.timetable['freq']) + self.timetable['freq']
116 # draw performance graph
117 run_time = str(int((t_end - t_start) * 1000))
118 #run_time_chart = "BEGIN netdata.plugin_pythond_" + self.chart_name + " " + str(since_last) + '\n'
119 #run_time_chart += "SET run_time = " + run_time + '\n'
120 #run_time_chart += "END\n"
121 #sys.stdout.write(run_time_chart)
122 sys.stdout.write("BEGIN netdata.plugin_pythond_%s %s\nSET run_time = %s\nEND\n" % \
123 (self.chart_name, str(since_last), run_time))
125 #msg.debug(self.chart_name + " updated in " + str(run_time) + " ms")
126 msg.debug(self.chart_name, "updated in", str(run_time), "ms")
127 self.timetable['last'] = t_start
132 Runs job in thread. Handles retries.
133 Exits when job failed or timed out.
136 self.timetable['last'] = time.time()
139 status = self._run_once()
140 except Exception as e:
141 msg.error("Something wrong: ", str(e))
144 time.sleep(self.timetable['next'] - time.time())
145 self.retries_left = self.retries
147 self.retries_left -= 1
148 if self.retries_left <= 0:
149 msg.error("no more retries. Exiting")
152 time.sleep(self.timetable['freq'])
157 append = params.append
162 if type(p) is not str:
169 def _line(self, instruction, *params):
171 Converts *params to string and joins them with one space between every one.
172 :param params: str/int/float
174 #self._data_stream += instruction
175 tmp = list(map((lambda x: "''" if x is None or len(x) == 0 else x), params))
177 self._data_stream += "%s %s\n" % (instruction, str(" ".join(tmp)))
179 # self.error(str(" ".join(tmp)))
189 # self._data_stream += " " + p
190 #self._data_stream += "\n"
192 def chart(self, type_id, name="", title="", units="", family="",
193 category="", charttype="line", priority="", update_every=""):
202 :param charttype: str
203 :param priority: int/str
204 :param update_every: int/str
206 self._charts.append(type_id)
207 #self._line("CHART", type_id, name, title, units, family, category, charttype, priority, update_every)
209 p = self._format(type_id, name, title, units, family, category, charttype, priority, update_every)
210 self._line("CHART", *p)
212 def dimension(self, id, name=None, algorithm="absolute", multiplier=1, divisor=1, hidden=False):
214 Defines a new dimension for the chart
217 :param algorithm: str
218 :param multiplier: int/str
219 :param divisor: int/str
220 :param hidden: boolean
226 self.error("malformed dimension: multiplier is not a number:", multiplier)
231 self.error("malformed dimension: divisor is not a number:", divisor)
235 if algorithm not in ("absolute", "incremental", "percentage-of-absolute-row", "percentage-of-incremental-row"):
236 algorithm = "absolute"
238 self._dimensions.append(str(id))
240 p = self._format(id, name, algorithm, multiplier, divisor, "hidden")
241 #self._line("DIMENSION", id, name, algorithm, str(multiplier), str(divisor), "hidden")
243 p = self._format(id, name, algorithm, multiplier, divisor)
244 #self._line("DIMENSION", id, name, algorithm, str(multiplier), str(divisor))
246 self._line("DIMENSION", *p)
248 def begin(self, type_id, microseconds=0):
252 :param microseconds: int
255 if type_id not in self._charts:
256 self.error("wrong chart type_id:", type_id)
261 self.error("malformed begin statement: microseconds are not a number:", microseconds)
264 self._line("BEGIN", type_id, str(microseconds))
267 def set(self, id, value):
269 Set value to dimension
271 :param value: int/float
274 if id not in self._dimensions:
275 self.error("wrong dimension id:", id, "Available dimensions are:", *self._dimensions)
278 value = str(int(value))
280 self.error("cannot set non-numeric value:", value)
282 self._line("SET", id, "=", str(value))
290 Upload new data to netdata
292 print(self._data_stream)
293 self._data_stream = ""
295 def error(self, *params):
297 Show error message on stderr
299 msg.error(self.chart_name, *params)
301 def debug(self, *params):
303 Show debug message on stderr
305 msg.debug(self.chart_name, *params)
307 def info(self, *params):
309 Show information message on stderr
311 msg.info(self.chart_name, *params)
318 msg.error("Service " + str(self.__module__) + "doesn't implement check() function")
326 msg.error("Service " + str(self.__module__) + "doesn't implement create() function?")
329 def update(self, interval):
335 msg.error("Service " + str(self.__module__) + "doesn't implement update() function")
339 class SimpleService(BaseService):
340 def __init__(self, configuration=None, name=None):
342 self.definitions = {}
343 BaseService.__init__(self, configuration=configuration, name=name)
363 data = self._get_data()
368 for name in self.order:
369 options = self.definitions[name]['options'] + [self.priority + idx, self.update_every]
370 self.chart(self.chart_name + "." + name, *options)
371 # check if server has this datapoint
372 for line in self.definitions[name]['lines']:
374 self.dimension(*line)
380 def update(self, interval):
386 data = self._get_data()
388 self.debug("_get_data() returned no data")
392 for chart in self.order:
393 if self.begin(self.chart_name + "." + chart, interval):
395 for dim in self.definitions[chart]['lines']:
397 self.set(dim[0], data[dim[0]])
404 self.error("no charts to update")
409 class UrlService(SimpleService):
410 def __init__(self, configuration=None, name=None):
414 SimpleService.__init__(self, configuration=configuration, name=name)
416 def __add_auth(self):
417 passman = urllib2.HTTPPasswordMgrWithDefaultRealm()
418 passman.add_password(None, self.url, self.user, self.password)
419 authhandler = urllib2.HTTPBasicAuthHandler(passman)
420 opener = urllib2.build_opener(authhandler)
421 urllib2.install_opener(opener)
423 def _get_raw_data(self):
425 Get raw data from http request
430 f = urllib2.urlopen(self.url, timeout=self.update_every)
431 except Exception as e:
436 raw = f.read().decode('utf-8')
437 except Exception as e:
445 Format configuration data and try to connect to server
448 if self.name is None or self.name == str(None):
450 self.chart_name += "_" + self.name
452 self.name = str(self.name)
454 self.url = str(self.configuration['url'])
455 except (KeyError, TypeError):
458 self.user = str(self.configuration['user'])
459 except (KeyError, TypeError):
462 self.password = str(self.configuration['pass'])
463 except (KeyError, TypeError):
466 if self.user is not None and self.password is not None:
469 if self._get_data() is not None:
475 class SocketService(SimpleService):
476 def __init__(self, configuration=None, name=None):
478 self._keep_alive = True
479 self.host = "localhost"
481 self.unix_socket = None
483 self.__socket_config = None
484 SimpleService.__init__(self, configuration=configuration, name=name)
488 Recreate socket and connect to it since they cannot be reused
489 Available configurations are IPv6, IPv4 or UNIX socket
493 if self.unix_socket is None:
494 if self.__socket_config is None:
495 # establish ipv6 or ipv4 connection.
496 for res in socket.getaddrinfo(self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM):
498 af, socktype, proto, canonname, sa = res
499 self._sock = socket.socket(af, socktype, proto)
500 except socket.error as msg:
504 self._sock.connect(sa)
505 except socket.error as msg:
508 self.__socket_config = res
511 # connect to socket with previously established configuration
513 af, socktype, proto, canonname, sa = self.__socket_config
514 self._sock = socket.socket(af, socktype, proto)
515 self._sock.connect(sa)
516 except socket.error as msg:
519 # connect to unix socket
520 self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
521 self._sock.connect(self.unix_socket)
522 except Exception as e:
524 "Cannot create socket with following configuration: host:", str(self.host),
525 "port:", str(self.port),
526 "socket:", str(self.unix_socket))
528 self._sock.setblocking(0)
530 def _disconnect(self):
532 Close socket connection
536 self._sock.shutdown(2) # 0 - read, 1 - write, 2 - all
547 # Send request if it is needed
548 if self.request != "".encode():
550 self._sock.send(self.request)
551 except Exception as e:
554 "used configuration: host:", str(self.host),
555 "port:", str(self.port),
556 "socket:", str(self.unix_socket))
562 Receive data from socket
568 ready_to_read, _, in_error = select.select([self._sock], [], [], 0.01)
569 except Exception as e:
570 self.debug("SELECT", str(e))
573 if len(ready_to_read) > 0:
574 buf = self._sock.recv(4096)
575 if len(buf) == 0 or buf is None:
584 def _get_raw_data(self):
586 Get raw data with low-level "socket" module.
589 if self._sock is None:
592 # Send request if it is needed
598 prevent_infinite_loop = 1000000
600 data += self._receive()
601 finished = self._check_raw_data(data)
602 prevent_infinite_loop -= 1
603 if prevent_infinite_loop <= 0:
604 self.debug("Almost got into infinite loop while grabbing data. Is _check_raw_data() ok?")
607 if not self._keep_alive:
612 def _check_raw_data(self, data):
614 Check if all data has been gathered from socket
620 def _parse_config(self):
622 Parse configuration data
625 if self.name is None or self.name == str(None):
628 self.name = str(self.name)
630 self.unix_socket = str(self.configuration['socket'])
631 except (KeyError, TypeError):
632 self.debug("No unix socket specified. Trying TCP/IP socket.")
634 self.host = str(self.configuration['host'])
635 except (KeyError, TypeError):
636 self.debug("No host specified. Using: '" + self.host + "'")
638 self.port = int(self.configuration['port'])
639 except (KeyError, TypeError):
640 self.debug("No port specified. Using: '" + str(self.port) + "'")
642 self.request = str(self.configuration['request'])
643 except (KeyError, TypeError):
644 self.debug("No request specified. Using: '" + str(self.request) + "'")
645 self.request = self.request.encode()
648 class LogService(SimpleService):
649 def __init__(self, configuration=None, name=None):
651 self._last_position = 0
652 # self._log_reader = None
653 SimpleService.__init__(self, configuration=configuration, name=name)
654 self.retries = 100000 # basically always retry
656 def _get_raw_data(self):
658 Get log lines since last poll
663 if os.path.getsize(self.log_path) < self._last_position:
664 self._last_position = 0
665 elif os.path.getsize(self.log_path) == self._last_position:
666 self.debug("Log file hasn't changed. No new data.")
668 with open(self.log_path, "r") as fp:
669 fp.seek(self._last_position)
670 for i, line in enumerate(fp):
672 self._last_position = fp.tell()
673 except Exception as e:
679 self.error("No data collected.")
684 Parse basic configuration and check if log file exists
687 if self.name is not None or self.name != str(None):
690 self.name = str(self.name)
692 self.log_path = str(self.configuration['path'])
693 except (KeyError, TypeError):
694 self.error("No path to log specified. Using: '" + self.log_path + "'")
696 if os.access(self.log_path, os.R_OK):
699 self.error("Cannot access file: '" + self.log_path + "'")
703 status = SimpleService.create(self)
704 self._last_position = 0
708 class ExecutableService(SimpleService):
709 #command_whitelist = ['exim', 'postqueue']
710 bad_substrings = ('&', '|', ';', '>', '<')
712 def __init__(self, configuration=None, name=None):
714 SimpleService.__init__(self, configuration=configuration, name=name)
716 def _get_raw_data(self):
718 Get raw data from executed command
722 p = Popen(self.command, stdout=PIPE, stderr=PIPE)
723 except Exception as e:
724 self.error("Executing command", self.command, "resulted in error:", str(e))
727 for line in p.stdout.readlines():
728 data.append(str(line.decode()))
731 self.error("No data collected.")
738 Parse basic configuration, check if command is whitelisted and is returning values
741 if self.name is not None or self.name != str(None):
744 self.name = str(self.name)
746 self.command = str(self.configuration['command'])
747 except (KeyError, TypeError):
748 self.error("No command specified. Using: '" + self.command + "'")
749 self.command = self.command.split(' ')
750 #if self.command[0] not in self.command_whitelist:
751 # self.error("Command is not whitelisted.")
754 for arg in self.command[1:]:
755 if any(st in arg for st in self.bad_substrings):
756 self.error("Bad command argument:" + " ".join(self.command[1:]))
758 # test command and search for it in /usr/sbin or /sbin when failed
759 base = self.command[0].split('/')[-1]
760 if self._get_raw_data() is None:
761 for prefix in ['/sbin/', '/usr/sbin/']:
762 self.command[0] = prefix + base
763 if os.path.isfile(self.command[0]):
765 #if self._get_raw_data() is not None:
768 if self._get_data() is None or len(self._get_data()) == 0:
769 self.error("Command", self.command, "returned no data")