diff options
Diffstat (limited to 'python.d/python_modules/base.py')
| -rw-r--r-- | python.d/python_modules/base.py | 297 |
1 files changed, 194 insertions, 103 deletions
diff --git a/python.d/python_modules/base.py b/python.d/python_modules/base.py index 1508e0965..320c54bae 100644 --- a/python.d/python_modules/base.py +++ b/python.d/python_modules/base.py @@ -116,33 +116,21 @@ class SimpleService(threading.Thread): Return value presents exit status of update() :return: boolean """ - t_start = time.time() - timetable = self.timetable + t_start = float(time.time()) chart_name = self.chart_name - # check if it is time to execute job update() function - if timetable['next'] > t_start: - self.debug(chart_name, "will be run in", str(int((timetable['next'] - t_start) * 1000)), "ms") - return True - since_last = int((t_start - timetable['last']) * 1000000) - self.debug(chart_name, - "ready to run, after", str(int((t_start - timetable['last']) * 1000)), - "ms (update_every:", str(timetable['freq'] * 1000), - "ms, latency:", str(int((t_start - timetable['next']) * 1000)), "ms") + 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 - t_end = time.time() - self.timetable['next'] = t_end - (t_end % timetable['freq']) + timetable['freq'] + # draw performance graph - run_time = str(int((t_end - t_start) * 1000)) - # noinspection SqlNoDataSourceInspection + 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), run_time)) - # sys.stdout.write("BEGIN netdata.plugin_pythond_%s %s\nSET run_time = %s\nEND\n" % - # (self.chart_name, str(since_last), run_time)) + (self.chart_name, str(since_last), str(run_time))) self.debug(chart_name, "updated in", str(run_time), "ms") self.timetable['last'] = t_start @@ -155,23 +143,48 @@ class SimpleService(threading.Thread): Exits when job failed or timed out. :return: None """ - self.timetable['last'] = time.time() + 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 as e: - self.error("Something wrong: ", str(e)) - return - if status: # handle retries if update failed - time.sleep(self.timetable['next'] - time.time()) + 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 <= 0: - self.error("no more retries. Exiting") - return + if penalty == 0: + penalty = float(self.retries * step) / 2 + else: + penalty *= 1.5 + + if penalty > 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: - time.sleep(self.timetable['freq']) + self.error("failed to collect data - " + str(self.retries_left) + " retries left - penalty: " + str(penalty) + " sec") # --- CHART --- @@ -307,7 +320,10 @@ class SimpleService(threading.Thread): """ Upload new data to netdata. """ - print(self._data_stream) + try: + print(self._data_stream) + except Exception as e: + msg.fatal('cannot send data to netdata:', str(e)) self._data_stream = "" # --- ERROR HANDLING --- @@ -318,6 +334,12 @@ class SimpleService(threading.Thread): """ 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 @@ -345,10 +367,18 @@ class SimpleService(threading.Thread): :return: boolean """ self.debug("Module", str(self.__module__), "doesn't implement check() function. Using default.") - if self._get_data() is None or len(self._get_data()) == 0: + data = self._get_data() + + if data is None: + self.debug("failed to receive data during check().") return False - else: - return True + + 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): """ @@ -357,6 +387,7 @@ class SimpleService(threading.Thread): """ data = self._get_data() if data is None: + self.debug("failed to receive data during create().") return False idx = 0 @@ -380,7 +411,7 @@ class SimpleService(threading.Thread): """ data = self._get_data() if data is None: - self.debug("_get_data() returned no data") + self.debug("failed to receive data during update().") return False updated = False @@ -458,7 +489,7 @@ class UrlService(SimpleService): return None try: - raw = f.read().decode('utf-8') + raw = f.read().decode('utf-8', 'ignore') except Exception as e: self.error(str(e)) finally: @@ -506,8 +537,83 @@ class SocketService(SimpleService): 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 @@ -515,63 +621,38 @@ class SocketService(SimpleService): :return: """ try: - if self.unix_socket is None: - if self.__socket_config is None: - # establish ipv6 or ipv4 connection. - for res in socket.getaddrinfo(self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM): - try: - # noinspection SpellCheckingInspection - af, socktype, proto, canonname, sa = res - self._sock = socket.socket(af, socktype, proto) - except socket.error as e: - self.debug("Cannot create socket:", str(e)) - self._sock = None - continue - try: - self._sock.connect(sa) - except socket.error as e: - self.debug("Cannot connect to socket:", str(e)) - self._disconnect() - continue - self.__socket_config = res - break - else: - # connect to socket with previously established configuration - try: - af, socktype, proto, canonname, sa = self.__socket_config - self._sock = socket.socket(af, socktype, proto) - self._sock.connect(sa) - except socket.error as e: - self.debug("Cannot create or connect to socket:", str(e)) - self._disconnect() + if self.unix_socket is not None: + self._connect2unixsocket() + else: - # connect to unix socket - try: - self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) - self._sock.connect(self.unix_socket) - except socket.error: - self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - self._sock.connect(self.unix_socket) + 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.error(str(e), - "Cannot create socket with following configuration: host:", str(self.host), - "port:", str(self.port), - "socket:", str(self.unix_socket)) self._sock = None - self._sock.setblocking(0) + 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: """ - try: - self._sock.shutdown(2) # 0 - read, 1 - write, 2 - all - self._sock.close() - except Exception: - pass - self._sock = None + 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): """ @@ -579,15 +660,13 @@ class SocketService(SimpleService): :return: boolean """ # Send request if it is needed - if self.request != "".encode(): + 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() - self.error(str(e), - "used configuration: host:", str(self.host), - "port:", str(self.port), - "socket:", str(self.unix_socket)) return False return True @@ -598,24 +677,28 @@ class SocketService(SimpleService): """ data = "" while True: + self.debug("receiving response") try: - ready_to_read, _, in_error = select.select([self._sock], [], [], 5) + buf = self._sock.recv(4096) except Exception as e: - self.debug("SELECT", str(e)) + self._socketerror("failed to receive response:" + str(e)) self._disconnect() break - if len(ready_to_read) > 0: - buf = self._sock.recv(4096) - if len(buf) == 0 or buf is None: # handle server disconnect - break - data += buf.decode() - if self._check_raw_data(data): - break - else: - self.error("Socket timed out.") + + 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): @@ -625,6 +708,8 @@ class SocketService(SimpleService): """ if self._sock is None: self._connect() + if self._sock is None: + return None # Send request if it is needed if not self._send(): @@ -654,6 +739,7 @@ class SocketService(SimpleService): self.name = "" else: self.name = str(self.name) + try: self.unix_socket = str(self.configuration['socket']) except (KeyError, TypeError): @@ -667,10 +753,12 @@ class SocketService(SimpleService): 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): @@ -724,7 +812,7 @@ class LogService(SimpleService): try: self.log_path = str(self.configuration['path']) except (KeyError, TypeError): - self.error("No path to log specified. Using: '" + self.log_path + "'") + self.info("No path to log specified. Using: '" + self.log_path + "'") if os.access(self.log_path, os.R_OK): return True @@ -779,22 +867,25 @@ class ExecutableService(SimpleService): try: self.command = str(self.configuration['command']) except (KeyError, TypeError): - self.error("No command specified. Using: '" + self.command + "'") - command = self.command.split(' ') + self.info("No command specified. Using: '" + self.command + "'") + # Splitting self.command on every space so subprocess.Popen reads it properly + self.command = self.command.split(' ') - for arg in command[1:]: + for arg in self.command[1:]: if any(st in arg for st in self.bad_substrings): self.error("Bad command argument:" + " ".join(self.command[1:])) return False + # test command and search for it in /usr/sbin or /sbin when failed - base = command[0].split('/')[-1] + base = self.command[0].split('/')[-1] if self._get_raw_data() is None: for prefix in ['/sbin/', '/usr/sbin/']: - command[0] = prefix + base - if os.path.isfile(command[0]): + self.command[0] = prefix + base + if os.path.isfile(self.command[0]): break - self.command = command + if self._get_data() is None or len(self._get_data()) == 0: self.error("Command", self.command, "returned no data") return False + return True |