python.d/python_modules/base.py

   1 # -*- coding: utf-8 -*-
   2 # Description: prototypes for netdata python.d modules
   3 # Author: Pawel Krupa (paulfantom)
   4
   5 import time
   6 import sys
   7 import os
   8 import socket
   9 try:
  10     import urllib.request as urllib2
  11 except ImportError:
  12     import urllib2
  13
  14 from subprocess import Popen, PIPE
  15
  16 import threading
  17 import msg
  18
  19
  20 class BaseService(threading.Thread):
  21     """
  22     Prototype of Service class.
  23     Implemented basic functionality to run jobs by `python.d.plugin`
  24     """
  25     def __init__(self, configuration=None, name=None):
  26         """
  27         This needs to be initialized in child classes
  28         :param configuration: dict
  29         :param name: str
  30         """
  31         threading.Thread.__init__(self)
  32         self._data_stream = ""
  33         self.daemon = True
  34         self.retries = 0
  35         self.retries_left = 0
  36         self.priority = 140000
  37         self.update_every = 1
  38         self.name = name
  39         self.override_name = None
  40         self.chart_name = ""
  41         self._dimensions = []
  42         self._charts = []
  43         if configuration is None:
  44             self.error("BaseService: no configuration parameters supplied. Cannot create Service.")
  45             raise RuntimeError
  46         else:
  47             self._extract_base_config(configuration)
  48             self.timetable = {}
  49             self.create_timetable()
  50
  51     def _extract_base_config(self, config):
  52         """
  53         Get basic parameters to run service
  54         Minimum config:
  55             config = {'update_every':1,
  56                       'priority':100000,
  57                       'retries':0}
  58         :param config: dict
  59         """
  60         try:
  61             self.override_name = config.pop('name')
  62         except KeyError:
  63             pass
  64         self.update_every = int(config.pop('update_every'))
  65         self.priority = int(config.pop('priority'))
  66         self.retries = int(config.pop('retries'))
  67         self.retries_left = self.retries
  68         self.configuration = config
  69
  70     def create_timetable(self, freq=None):
  71         """
  72         Create service timetable.
  73         `freq` is optional
  74         Example:
  75             timetable = {'last': 1466370091.3767564,
  76                          'next': 1466370092,
  77                          'freq': 1}
  78         :param freq: int
  79         """
  80         if freq is None:
  81             freq = self.update_every
  82         now = time.time()
  83         self.timetable = {'last': now,
  84                           'next': now - (now % freq) + freq,
  85                           'freq': freq}
  86
  87     def _run_once(self):
  88         """
  89         Executes self.update(interval) and draws run time chart.
  90         Return value presents exit status of update()
  91         :return: boolean
  92         """
  93         t_start = time.time()
  94         # check if it is time to execute job update() function
  95         if self.timetable['next'] > t_start:
  96             msg.debug(self.chart_name + " will be run in " +
  97                       str(int((self.timetable['next'] - t_start) * 1000)) + " ms")
  98             return True
  99
 100         since_last = int((t_start - self.timetable['last']) * 1000000)
 101         msg.debug(self.chart_name +
 102                   " ready to run, after " + str(int((t_start - self.timetable['last']) * 1000)) +
 103                   " ms (update_every: " + str(self.timetable['freq'] * 1000) +
 104                   " ms, latency: " + str(int((t_start - self.timetable['next']) * 1000)) + " ms)")
 105         if not self.update(since_last):
 106             return False
 107         t_end = time.time()
 108         self.timetable['next'] = t_end - (t_end % self.timetable['freq']) + self.timetable['freq']
 109
 110         # draw performance graph
 111         run_time = str(int((t_end - t_start) * 1000))
 112         run_time_chart = "BEGIN netdata.plugin_pythond_" + self.chart_name + " " + str(since_last) + '\n'
 113         run_time_chart += "SET run_time = " + run_time + '\n'
 114         run_time_chart += "END\n"
 115         sys.stdout.write(run_time_chart)
 116         msg.debug(self.chart_name + " updated in " + str(run_time) + " ms")
 117         self.timetable['last'] = t_start
 118         return True
 119
 120     def run(self):
 121         """
 122         Runs job in thread. Handles retries.
 123         Exits when job failed or timed out.
 124         :return: None
 125         """
 126         self.timetable['last'] = time.time()
 127         while True:
 128             try:
 129                 status = self._run_once()
 130             except Exception as e:
 131                 msg.error("Something wrong: " + str(e))
 132                 return
 133             if status:
 134                 time.sleep(self.timetable['next'] - time.time())
 135                 self.retries_left = self.retries
 136             else:
 137                 self.retries_left -= 1
 138                 if self.retries_left <= 0:
 139                     msg.error("no more retries. Exiting")
 140                     return
 141                 else:
 142                     time.sleep(self.timetable['freq'])
 143
 144     def _line(self, instruction, *params):
 145         """
 146         Converts *params to string and joins them with one space between every one.
 147         :param params: str/int/float
 148         """
 149         self._data_stream += instruction
 150         for p in params:
 151             if p is None:
 152                 p = ""
 153             else:
 154                 p = str(p)
 155             if len(p) == 0:
 156                 p = "''"
 157             if ' ' in p:
 158                 p = "'" + p + "'"
 159             self._data_stream += " " + p
 160         self._data_stream += "\n"
 161
 162     def chart(self, type_id, name="", title="", units="", family="",
 163               category="", charttype="line", priority="", update_every=""):
 164         """
 165         Defines a new chart.
 166         :param type_id: str
 167         :param name: str
 168         :param title: str
 169         :param units: str
 170         :param family: str
 171         :param category: str
 172         :param charttype: str
 173         :param priority: int/str
 174         :param update_every: int/str
 175         """
 176         self._charts.append(type_id)
 177         self._line("CHART", type_id, name, title, units, family, category, charttype, priority, update_every)
 178
 179     def dimension(self, id, name=None, algorithm="absolute", multiplier=1, divisor=1, hidden=False):
 180         """
 181         Defines a new dimension for the chart
 182         :param id: str
 183         :param name: str
 184         :param algorithm: str
 185         :param multiplier: int/str
 186         :param divisor: int/str
 187         :param hidden: boolean
 188         :return:
 189         """
 190         try:
 191             int(multiplier)
 192         except TypeError:
 193             self.error("malformed dimension: multiplier is not a number:", multiplier)
 194             multiplier = 1
 195         try:
 196             int(divisor)
 197         except TypeError:
 198             self.error("malformed dimension: divisor is not a number:", divisor)
 199             divisor = 1
 200         if name is None:
 201             name = id
 202         if algorithm not in ("absolute", "incremental", "percentage-of-absolute-row", "percentage-of-incremental-row"):
 203             algorithm = "absolute"
 204
 205         self._dimensions.append(id)
 206         if hidden:
 207             self._line("DIMENSION", id, name, algorithm, multiplier, divisor, "hidden")
 208         else:
 209             self._line("DIMENSION", id, name, algorithm, multiplier, divisor)
 210
 211     def begin(self, type_id, microseconds=0):
 212         """
 213         Begin data set
 214         :param type_id: str
 215         :param microseconds: int
 216         :return: boolean
 217         """
 218         if type_id not in self._charts:
 219             self.error("wrong chart type_id:", type_id)
 220             return False
 221         try:
 222             int(microseconds)
 223         except TypeError:
 224             self.error("malformed begin statement: microseconds are not a number:", microseconds)
 225             microseconds = ""
 226
 227         self._line("BEGIN", type_id, microseconds)
 228         return True
 229
 230     def set(self, id, value):
 231         """
 232         Set value to dimension
 233         :param id: str
 234         :param value: int/float
 235         :return: boolean
 236         """
 237         if id not in self._dimensions:
 238             self.error("wrong dimension id:", id)
 239             return False
 240         try:
 241             value = str(int(value))
 242         except TypeError:
 243             self.error("cannot set non-numeric value:", value)
 244             return False
 245         self._line("SET", id, "=", value)
 246         return True
 247
 248     def end(self):
 249         self._line("END")
 250
 251     def commit(self):
 252         """
 253         Upload new data to netdata
 254         """
 255         print(self._data_stream)
 256         self._data_stream = ""
 257
 258     def error(self, *params):
 259         """
 260         Show error message on stderr
 261         """
 262         msg.error(self.chart_name, *params)
 263
 264     def debug(self, *params):
 265         """
 266         Show debug message on stderr
 267         """
 268         msg.debug(self.chart_name, *params)
 269
 270     def info(self, *params):
 271         """
 272         Show information message on stderr
 273         """
 274         msg.info(self.chart_name, *params)
 275
 276     def check(self):
 277         """
 278         check() prototype
 279         :return: boolean
 280         """
 281         msg.error("Service " + str(self.__module__) + "doesn't implement check() function")
 282         return False
 283
 284     def create(self):
 285         """
 286         create() prototype
 287         :return: boolean
 288         """
 289         msg.error("Service " + str(self.__module__) + "doesn't implement create() function?")
 290         return False
 291
 292     def update(self, interval):
 293         """
 294         update() prototype
 295         :param interval: int
 296         :return: boolean
 297         """
 298         msg.error("Service " + str(self.__module__) + "doesn't implement update() function")
 299         return False
 300
 301
 302 class SimpleService(BaseService):
 303     def __init__(self, configuration=None, name=None):
 304         self.order = []
 305         self.definitions = {}
 306         BaseService.__init__(self, configuration=configuration, name=name)
 307
 308     def _get_data(self):
 309         """
 310         Get some data
 311         :return: dict
 312         """
 313         return {}
 314
 315     def check(self):
 316         """
 317         :return:
 318         """
 319         return True
 320
 321     def create(self):
 322         """
 323         Create charts
 324         :return: boolean
 325         """
 326         data = self._get_data()
 327         if data is None:
 328             return False
 329
 330         idx = 0
 331         for name in self.order:
 332             options = self.definitions[name]['options'] + [self.priority + idx, self.update_every]
 333             self.chart(self.chart_name + "." + name, *options)
 334             # check if server has this datapoint
 335             for line in self.definitions[name]['lines']:
 336                 if line[0] in data:
 337                     self.dimension(*line)
 338             idx += 1
 339
 340         self.commit()
 341         return True
 342
 343     def update(self, interval):
 344         """
 345         Update charts
 346         :param interval: int
 347         :return: boolean
 348         """
 349         data = self._get_data()
 350         if data is None:
 351             return False
 352
 353         updated = False
 354         for chart in self.order:
 355             if self.begin(self.chart_name + "." + chart, interval):
 356                 updated = True
 357                 for dim in self.definitions[chart]['lines']:
 358                     try:
 359                         self.set(dim[0], data[dim[0]])
 360                     except KeyError:
 361                         pass
 362                 self.end()
 363
 364         self.commit()
 365
 366         return updated
 367
 368
 369 class UrlService(SimpleService):
 370     def __init__(self, configuration=None, name=None):
 371         self.url = ""
 372         self.user = None
 373         self.password = None
 374         SimpleService.__init__(self, configuration=configuration, name=name)
 375
 376     def __add_auth(self):
 377         passman = urllib2.HTTPPasswordMgrWithDefaultRealm()
 378         passman.add_password(None, self.url, self.user, self.password)
 379         authhandler = urllib2.HTTPBasicAuthHandler(passman)
 380         opener = urllib2.build_opener(authhandler)
 381         urllib2.install_opener(opener)
 382
 383     def _get_raw_data(self):
 384         """
 385         Get raw data from http request
 386         :return: str
 387         """
 388         raw = None
 389         try:
 390             f = urllib2.urlopen(self.url, timeout=self.update_every)
 391         except Exception as e:
 392             msg.error(self.__module__, str(e))
 393             return None
 394
 395         try:
 396             raw = f.read().decode('utf-8')
 397         except Exception as e:
 398             msg.error(self.__module__, str(e))
 399         finally:
 400             f.close()
 401         return raw
 402
 403     def check(self):
 404         """
 405         Format configuration data and try to connect to server
 406         :return: boolean
 407         """
 408         if self.name is None or self.name == str(None):
 409             self.name = 'local'
 410             self.chart_name += "_" + self.name
 411         else:
 412             self.name = str(self.name)
 413         try:
 414             self.url = str(self.configuration['url'])
 415         except (KeyError, TypeError):
 416             pass
 417         try:
 418             self.user = str(self.configuration['user'])
 419         except (KeyError, TypeError):
 420             pass
 421         try:
 422             self.password = str(self.configuration['password'])
 423         except (KeyError, TypeError):
 424             pass
 425
 426         if self.user is not None and self.password is not None:
 427             self.__add_auth()
 428
 429         if self._get_data() is not None:
 430             return True
 431         else:
 432             return False
 433
 434
 435 class SocketService(SimpleService):
 436     def __init__(self, configuration=None, name=None):
 437         self.host = "localhost"
 438         self.port = None
 439         self.sock = None
 440         self.unix_socket = None
 441         self.request = ""
 442         SimpleService.__init__(self, configuration=configuration, name=name)
 443
 444     def _get_raw_data(self):
 445         """
 446         Get raw data with low-level "socket" module.
 447         :return: str
 448         """
 449         if self.sock is None:
 450             try:
 451                 if self.unix_socket is None:
 452                     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 453                     sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 454                     sock.settimeout(self.update_every)
 455                     sock.connect((self.host, self.port))
 456                 else:
 457                     sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
 458                     sock.settimeout(self.update_every)
 459                     sock.connect(self.unix_socket)
 460
 461             except Exception as e:
 462                 self.error(e)
 463                 self.sock = None
 464                 return None
 465
 466         if self.request != "".encode():
 467             try:
 468                 sock.send(self.request)
 469             except Exception:
 470                 try:
 471                     sock.shutdown(1)
 472                     sock.close()
 473                 except:
 474                     pass
 475                 self.sock = None
 476                 return None
 477
 478         data = sock.recv(2)
 479         try:
 480             while True:
 481                 buf = sock.recv(1024)
 482 #                if not buf:
 483 #                    break
 484 #                else:
 485 #                    data += buf
 486                 data += buf
 487                 if len(buf) < 1024:
 488                     break
 489         except:
 490             sock.close()
 491             return None
 492
 493         return data.decode()
 494
 495     def _parse_config(self):
 496         """
 497         Parse configuration data
 498         :return: boolean
 499         """
 500         if self.name is not None or self.name != str(None):
 501             self.name = ""
 502         else:
 503             self.name = str(self.name)
 504         try:
 505             self.unix_socket = int(self.configuration['socket'])
 506         except (KeyError, TypeError):
 507             self.error("No unix socket specified. Trying TCP/IP socket.")
 508             try:
 509                 self.host = str(self.configuration['host'])
 510             except (KeyError, TypeError):
 511                 self.error("No host specified. Using: '" + self.host + "'")
 512             try:
 513                 self.port = int(self.configuration['port'])
 514             except (KeyError, TypeError):
 515                 self.error("No port specified. Using: '" + str(self.port) + "'")
 516         try:
 517             self.request = str(self.configuration['request'])
 518         except (KeyError, TypeError):
 519             self.error("No request specified. Using: '" + str(self.request) + "'")
 520         self.request = self.request.encode()
 521
 522
 523 class LogService(SimpleService):
 524     def __init__(self, configuration=None, name=None):
 525         self.log_path = ""
 526         self._last_position = 0
 527         # self._log_reader = None
 528         SimpleService.__init__(self, configuration=configuration, name=name)
 529         self.retries = 100000  # basically always retry
 530
 531     def _get_raw_data(self):
 532         """
 533         Get log lines since last poll
 534         :return: list
 535         """
 536         lines = []
 537         try:
 538             if os.path.getsize(self.log_path) < self._last_position:
 539                 self._last_position = 0
 540             elif os.path.getsize(self.log_path) == self._last_position:
 541                 return None
 542             with open(self.log_path, "r") as fp:
 543                 fp.seek(self._last_position)
 544                 for i, line in enumerate(fp):
 545                     lines.append(line)
 546                 self._last_position = fp.tell()
 547         except Exception as e:
 548             self.error(self.__module__, str(e))
 549
 550         if len(lines) != 0:
 551             return lines
 552         return None
 553
 554     def check(self):
 555         """
 556         Parse basic configuration and check if log file exists
 557         :return: boolean
 558         """
 559         if self.name is not None or self.name != str(None):
 560             self.name = ""
 561         else:
 562             self.name = str(self.name)
 563         try:
 564             self.log_path = str(self.configuration['path'])
 565         except (KeyError, TypeError):
 566             self.error("No path to log specified. Using: '" + self.log_path + "'")
 567
 568         if os.access(self.log_path, os.R_OK):
 569             return True
 570         else:
 571             self.error("Cannot access file: '" + self.log_path + "'")
 572             return False
 573
 574     def create(self):
 575         status = SimpleService.create(self)
 576         self._last_position = 0
 577         return status
 578
 579
 580 class ExecutableService(SimpleService):
 581     command_whitelist = ['exim', 'postqueue']
 582     bad_substrings = ('&', '|', ';', '>', '<')
 583
 584     def __init__(self, configuration=None, name=None):
 585         self.command = ""
 586         SimpleService.__init__(self, configuration=configuration, name=name)
 587
 588     def _get_raw_data(self):
 589         """
 590         Get raw data from executed command
 591         :return: str
 592         """
 593         try:
 594             p = Popen(self.command, stdout=PIPE, stderr=PIPE)
 595         except Exception as e:
 596             self.error(self.__module__, str(e))
 597             return None
 598         data = []
 599         for line in p.stdout.readlines():
 600             data.append(str(line.decode()))
 601
 602         return data
 603
 604     def check(self):
 605         """
 606         Parse basic configuration, check if command is whitelisted and is returning values
 607         :return: boolean
 608         """
 609         if self.name is not None or self.name != str(None):
 610             self.name = ""
 611         else:
 612             self.name = str(self.name)
 613         # try:
 614         #     self.command = str(self.configuration['path'])
 615         # except (KeyError, TypeError):
 616         #     self.error("No command specified. Using: '" + self.command + "'")
 617         self.command = self.command.split(' ')
 618         if self.command[0] not in self.command_whitelist:
 619             self.error("Command is not whitelisted.")
 620             return False
 621
 622         for arg in self.command[1:]:
 623             if any(st in arg for st in self.bad_substrings):
 624                 self.error("Bad command argument:" + " ".join(self.command[1:]))
 625                 return False
 626         # test command and search for it in /usr/sbin or /sbin when failed
 627         base = self.command[0]
 628         if self._get_raw_data() is None:
 629             for prefix in ['/sbin/', '/usr/sbin/']:
 630                 self.command[0] = prefix + base
 631                 if os.path.isfile(self.command[0]):
 632                     break
 633                 #if self._get_raw_data() is not None:
 634                 #    break
 635
 636         if self._get_data() is None or len(self._get_data()) == 0:
 637             return False
 638         return True