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             msg.debug(self.chart_name, "will be run in", str(int((self.timetable['next'] - t_start) * 1000)), "ms")
  99             return True
 100
 101         since_last = int((t_start - self.timetable['last']) * 1000000)
 102         #msg.debug(self.chart_name +
 103         #          " ready to run, after " + str(int((t_start - self.timetable['last']) * 1000)) +
 104         #          " ms (update_every: " + str(self.timetable['freq'] * 1000) +
 105         #          " ms, latency: " + str(int((t_start - self.timetable['next']) * 1000)) + " ms)")
 106         msg.debug(self.chart_name,
 107                   "ready to run, after", str(int((t_start - self.timetable['last']) * 1000)),
 108                   "ms (update_every:", str(self.timetable['freq'] * 1000),
 109                   "ms, latency:", str(int((t_start - self.timetable['next']) * 1000)), "ms")
 110         if not self.update(since_last):
 111             self.error("update function failed.")
 112             return False
 113         t_end = time.time()
 114         self.timetable['next'] = t_end - (t_end % self.timetable['freq']) + self.timetable['freq']
 115         # draw performance graph
 116         run_time = str(int((t_end - t_start) * 1000))
 117         #run_time_chart = "BEGIN netdata.plugin_pythond_" + self.chart_name + " " + str(since_last) + '\n'
 118         #run_time_chart += "SET run_time = " + run_time + '\n'
 119         #run_time_chart += "END\n"
 120         #sys.stdout.write(run_time_chart)
 121         sys.stdout.write("BEGIN netdata.plugin_pythond_%s %s\nSET run_time = %s\nEND\n" % \
 122                          (self.chart_name, str(since_last), run_time))
 123
 124         #msg.debug(self.chart_name + " updated in " + str(run_time) + " ms")
 125         msg.debug(self.chart_name, "updated in", str(run_time), "ms")
 126         self.timetable['last'] = t_start
 127         return True
 128
 129     def run(self):
 130         """
 131         Runs job in thread. Handles retries.
 132         Exits when job failed or timed out.
 133         :return: None
 134         """
 135         self.timetable['last'] = time.time()
 136         while True:
 137             try:
 138                 status = self._run_once()
 139             except Exception as e:
 140                 msg.error("Something wrong: ", str(e))
 141                 return
 142             if status:
 143                 time.sleep(self.timetable['next'] - time.time())
 144                 self.retries_left = self.retries
 145             else:
 146                 self.retries_left -= 1
 147                 if self.retries_left <= 0:
 148                     msg.error("no more retries. Exiting")
 149                     return
 150                 else:
 151                     time.sleep(self.timetable['freq'])
 152
 153     @staticmethod
 154     def _format(*args):
 155         params = []
 156         append = params.append
 157         for p in args:
 158             if p is None:
 159                 append(p)
 160                 continue
 161             if type(p) is not str:
 162                 p = str(p)
 163             if ' ' in p:
 164                 p = "'" + p + "'"
 165             append(p)
 166         return params
 167
 168     def _line(self, instruction, *params):
 169         """
 170         Converts *params to string and joins them with one space between every one.
 171         :param params: str/int/float
 172         """
 173         #self._data_stream += instruction
 174         tmp = list(map((lambda x: "''" if x is None or len(x) == 0 else x), params))
 175
 176         self._data_stream += "%s %s\n" % (instruction, str(" ".join(tmp)))
 177
 178         # self.error(str(" ".join(tmp)))
 179         # for p in params:
 180         #     if p is None:
 181         #         p = ""
 182         #     else:
 183         #         p = str(p)
 184         #     if len(p) == 0:
 185         #         p = "''"
 186         #     if ' ' in p:
 187         #         p = "'" + p + "'"
 188         #     self._data_stream += " " + p
 189         #self._data_stream += "\n"
 190
 191     def chart(self, type_id, name="", title="", units="", family="",
 192               category="", charttype="line", priority="", update_every=""):
 193         """
 194         Defines a new chart.
 195         :param type_id: str
 196         :param name: str
 197         :param title: str
 198         :param units: str
 199         :param family: str
 200         :param category: str
 201         :param charttype: str
 202         :param priority: int/str
 203         :param update_every: int/str
 204         """
 205         self._charts.append(type_id)
 206         #self._line("CHART", type_id, name, title, units, family, category, charttype, priority, update_every)
 207
 208         p = self._format(type_id, name, title, units, family, category, charttype, priority, update_every)
 209         self._line("CHART", *p)
 210
 211     def dimension(self, id, name=None, algorithm="absolute", multiplier=1, divisor=1, hidden=False):
 212         """
 213         Defines a new dimension for the chart
 214         :param id: str
 215         :param name: str
 216         :param algorithm: str
 217         :param multiplier: int/str
 218         :param divisor: int/str
 219         :param hidden: boolean
 220         :return:
 221         """
 222         try:
 223             int(multiplier)
 224         except TypeError:
 225             self.error("malformed dimension: multiplier is not a number:", multiplier)
 226             multiplier = 1
 227         try:
 228             int(divisor)
 229         except TypeError:
 230             self.error("malformed dimension: divisor is not a number:", divisor)
 231             divisor = 1
 232         if name is None:
 233             name = id
 234         if algorithm not in ("absolute", "incremental", "percentage-of-absolute-row", "percentage-of-incremental-row"):
 235             algorithm = "absolute"
 236
 237         self._dimensions.append(str(id))
 238         if hidden:
 239             p = self._format(id, name, algorithm, multiplier, divisor, "hidden")
 240             #self._line("DIMENSION", id, name, algorithm, str(multiplier), str(divisor), "hidden")
 241         else:
 242             p = self._format(id, name, algorithm, multiplier, divisor)
 243             #self._line("DIMENSION", id, name, algorithm, str(multiplier), str(divisor))
 244
 245         self._line("DIMENSION", *p)
 246
 247     def begin(self, type_id, microseconds=0):
 248         """
 249         Begin data set
 250         :param type_id: str
 251         :param microseconds: int
 252         :return: boolean
 253         """
 254         if type_id not in self._charts:
 255             self.error("wrong chart type_id:", type_id)
 256             return False
 257         try:
 258             int(microseconds)
 259         except TypeError:
 260             self.error("malformed begin statement: microseconds are not a number:", microseconds)
 261             microseconds = ""
 262
 263         self._line("BEGIN", type_id, str(microseconds))
 264         return True
 265
 266     def set(self, id, value):
 267         """
 268         Set value to dimension
 269         :param id: str
 270         :param value: int/float
 271         :return: boolean
 272         """
 273         if id not in self._dimensions:
 274             self.error("wrong dimension id:", id, "Available dimensions are:", *self._dimensions)
 275             return False
 276         try:
 277             value = str(int(value))
 278         except TypeError:
 279             self.error("cannot set non-numeric value:", value)
 280             return False
 281         self._line("SET", id, "=", str(value))
 282         return True
 283
 284     def end(self):
 285         self._line("END")
 286
 287     def commit(self):
 288         """
 289         Upload new data to netdata
 290         """
 291         print(self._data_stream)
 292         self._data_stream = ""
 293
 294     def error(self, *params):
 295         """
 296         Show error message on stderr
 297         """
 298         msg.error(self.chart_name, *params)
 299
 300     def debug(self, *params):
 301         """
 302         Show debug message on stderr
 303         """
 304         msg.debug(self.chart_name, *params)
 305
 306     def info(self, *params):
 307         """
 308         Show information message on stderr
 309         """
 310         msg.info(self.chart_name, *params)
 311
 312     def check(self):
 313         """
 314         check() prototype
 315         :return: boolean
 316         """
 317         msg.error("Service " + str(self.__module__) + "doesn't implement check() function")
 318         return False
 319
 320     def create(self):
 321         """
 322         create() prototype
 323         :return: boolean
 324         """
 325         msg.error("Service " + str(self.__module__) + "doesn't implement create() function?")
 326         return False
 327
 328     def update(self, interval):
 329         """
 330         update() prototype
 331         :param interval: int
 332         :return: boolean
 333         """
 334         msg.error("Service " + str(self.__module__) + "doesn't implement update() function")
 335         return False
 336
 337
 338 class SimpleService(BaseService):
 339     def __init__(self, configuration=None, name=None):
 340         self.order = []
 341         self.definitions = {}
 342         BaseService.__init__(self, configuration=configuration, name=name)
 343
 344     def _get_data(self):
 345         """
 346         Get some data
 347         :return: dict
 348         """
 349         return {}
 350
 351     def check(self):
 352         """
 353         :return:
 354         """
 355         return True
 356
 357     def create(self):
 358         """
 359         Create charts
 360         :return: boolean
 361         """
 362         data = self._get_data()
 363         if data is None:
 364             return False
 365
 366         idx = 0
 367         for name in self.order:
 368             options = self.definitions[name]['options'] + [self.priority + idx, self.update_every]
 369             self.chart(self.chart_name + "." + name, *options)
 370             # check if server has this datapoint
 371             for line in self.definitions[name]['lines']:
 372                 if line[0] in data:
 373                     self.dimension(*line)
 374             idx += 1
 375
 376         self.commit()
 377         return True
 378
 379     def update(self, interval):
 380         """
 381         Update charts
 382         :param interval: int
 383         :return: boolean
 384         """
 385         data = self._get_data()
 386         if data is None:
 387             self.debug("_get_data() returned no data")
 388             return False
 389
 390         updated = False
 391         for chart in self.order:
 392             if self.begin(self.chart_name + "." + chart, interval):
 393                 updated = True
 394                 for dim in self.definitions[chart]['lines']:
 395                     try:
 396                         self.set(dim[0], data[dim[0]])
 397                     except KeyError:
 398                         pass
 399                 self.end()
 400
 401         self.commit()
 402         if not updated:
 403             self.error("no charts to update")
 404
 405         return updated
 406
 407
 408 class UrlService(SimpleService):
 409     def __init__(self, configuration=None, name=None):
 410         self.url = ""
 411         self.user = None
 412         self.password = None
 413         SimpleService.__init__(self, configuration=configuration, name=name)
 414
 415     def __add_auth(self):
 416         passman = urllib2.HTTPPasswordMgrWithDefaultRealm()
 417         passman.add_password(None, self.url, self.user, self.password)
 418         authhandler = urllib2.HTTPBasicAuthHandler(passman)
 419         opener = urllib2.build_opener(authhandler)
 420         urllib2.install_opener(opener)
 421
 422     def _get_raw_data(self):
 423         """
 424         Get raw data from http request
 425         :return: str
 426         """
 427         raw = None
 428         try:
 429             f = urllib2.urlopen(self.url, timeout=self.update_every)
 430         except Exception as e:
 431             self.error(str(e))
 432             return None
 433
 434         try:
 435             raw = f.read().decode('utf-8')
 436         except Exception as e:
 437             self.error(str(e))
 438         finally:
 439             f.close()
 440         return raw
 441
 442     def check(self):
 443         """
 444         Format configuration data and try to connect to server
 445         :return: boolean
 446         """
 447         if self.name is None or self.name == str(None):
 448             self.name = 'local'
 449             self.chart_name += "_" + self.name
 450         else:
 451             self.name = str(self.name)
 452         try:
 453             self.url = str(self.configuration['url'])
 454         except (KeyError, TypeError):
 455             pass
 456         try:
 457             self.user = str(self.configuration['user'])
 458         except (KeyError, TypeError):
 459             pass
 460         try:
 461             self.password = str(self.configuration['pass'])
 462         except (KeyError, TypeError):
 463             pass
 464
 465         if self.user is not None and self.password is not None:
 466             self.__add_auth()
 467
 468         if self._get_data() is not None:
 469             return True
 470         else:
 471             return False
 472
 473
 474 class SocketService(SimpleService):
 475     def __init__(self, configuration=None, name=None):
 476         self.host = "localhost"
 477         self.port = None
 478         self.sock = None
 479         self.unix_socket = None
 480         self.request = ""
 481         SimpleService.__init__(self, configuration=configuration, name=name)
 482
 483     def _get_raw_data(self):
 484         """
 485         Get raw data with low-level "socket" module.
 486         :return: str
 487         """
 488         if self.sock is None:
 489             try:
 490                 if self.unix_socket is None:
 491                     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 492                     sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 493                     #sock.setsockopt(socket.SOL_SOCKET, socket.TCP_NODELAY, 1)
 494                     #sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
 495                     #sock.settimeout(self.update_every)
 496                     sock.settimeout(0.5)
 497                     sock.connect((self.host, self.port))
 498                     sock.settimeout(0.5)  # Just to be sure
 499                 else:
 500                     sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
 501                     #sock.settimeout(self.update_every)
 502                     sock.settimeout(0.05)
 503                     sock.connect(self.unix_socket)
 504                     sock.settimeout(0.05)  # Just to be sure
 505
 506             except Exception as e:
 507                 self.error(str(e), "used configuration: host:", str(self.host), "port:", str(self.port), "socket:", str(self.unix_socket))
 508                 self.sock = None
 509                 return None
 510
 511         if self.request != "".encode():
 512             try:
 513                 sock.send(self.request)
 514             except Exception as e:
 515                 try:
 516                     sock.shutdown(1)
 517                     sock.close()
 518                 except:
 519                     pass
 520                 self.sock = None
 521                 self.error(str(e), "used configuration: host:", str(self.host), "port:", str(self.port), "socket:", str(self.unix_socket))
 522                 return None
 523
 524         size = 2
 525         try:
 526             data = sock.recv(size).decode()
 527         except Exception as e:
 528             self.error(str(e), "used configuration: host:", str(self.host), "port:", str(self.port), "socket:", str(self.unix_socket))
 529             sock.close()
 530             return None
 531
 532         while True:
 533             # implement something like TCP Window Scaling
 534             if size < 4096:
 535                 size *= 2
 536             buf = sock.recv(size)
 537             data += buf.decode()
 538             if len(buf) < size:
 539                 break
 540
 541         return data
 542
 543     def _parse_config(self):
 544         """
 545         Parse configuration data
 546         :return: boolean
 547         """
 548         if self.name is None or self.name == str(None):
 549             self.name = ""
 550         else:
 551             self.name = str(self.name)
 552         try:
 553             self.unix_socket = str(self.configuration['socket'])
 554         except (KeyError, TypeError):
 555             self.debug("No unix socket specified. Trying TCP/IP socket.")
 556             try:
 557                 self.host = str(self.configuration['host'])
 558             except (KeyError, TypeError):
 559                 self.debug("No host specified. Using: '" + self.host + "'")
 560             try:
 561                 self.port = int(self.configuration['port'])
 562             except (KeyError, TypeError):
 563                 self.debug("No port specified. Using: '" + str(self.port) + "'")
 564         try:
 565             self.request = str(self.configuration['request'])
 566         except (KeyError, TypeError):
 567             self.debug("No request specified. Using: '" + str(self.request) + "'")
 568         self.request = self.request.encode()
 569
 570
 571 class LogService(SimpleService):
 572     def __init__(self, configuration=None, name=None):
 573         self.log_path = ""
 574         self._last_position = 0
 575         # self._log_reader = None
 576         SimpleService.__init__(self, configuration=configuration, name=name)
 577         self.retries = 100000  # basically always retry
 578
 579     def _get_raw_data(self):
 580         """
 581         Get log lines since last poll
 582         :return: list
 583         """
 584         lines = []
 585         try:
 586             if os.path.getsize(self.log_path) < self._last_position:
 587                 self._last_position = 0
 588             elif os.path.getsize(self.log_path) == self._last_position:
 589                 self.debug("Log file hasn't changed. No new data.")
 590                 return None
 591             with open(self.log_path, "r") as fp:
 592                 fp.seek(self._last_position)
 593                 for i, line in enumerate(fp):
 594                     lines.append(line)
 595                 self._last_position = fp.tell()
 596         except Exception as e:
 597             self.error(str(e))
 598
 599         if len(lines) != 0:
 600             return lines
 601         else:
 602             self.error("No data collected.")
 603             return None
 604
 605     def check(self):
 606         """
 607         Parse basic configuration and check if log file exists
 608         :return: boolean
 609         """
 610         if self.name is not None or self.name != str(None):
 611             self.name = ""
 612         else:
 613             self.name = str(self.name)
 614         try:
 615             self.log_path = str(self.configuration['path'])
 616         except (KeyError, TypeError):
 617             self.error("No path to log specified. Using: '" + self.log_path + "'")
 618
 619         if os.access(self.log_path, os.R_OK):
 620             return True
 621         else:
 622             self.error("Cannot access file: '" + self.log_path + "'")
 623             return False
 624
 625     def create(self):
 626         status = SimpleService.create(self)
 627         self._last_position = 0
 628         return status
 629
 630
 631 class ExecutableService(SimpleService):
 632     #command_whitelist = ['exim', 'postqueue']
 633     bad_substrings = ('&', '|', ';', '>', '<')
 634
 635     def __init__(self, configuration=None, name=None):
 636         self.command = ""
 637         SimpleService.__init__(self, configuration=configuration, name=name)
 638
 639     def _get_raw_data(self):
 640         """
 641         Get raw data from executed command
 642         :return: str
 643         """
 644         try:
 645             p = Popen(self.command, stdout=PIPE, stderr=PIPE)
 646         except Exception as e:
 647             self.error(str(e))
 648             return None
 649         data = []
 650         for line in p.stdout.readlines():
 651             data.append(str(line.decode()))
 652
 653         if len(data) == 0:
 654             self.error("No data collected.")
 655             return None
 656
 657         return data
 658
 659     def check(self):
 660         """
 661         Parse basic configuration, check if command is whitelisted and is returning values
 662         :return: boolean
 663         """
 664         if self.name is not None or self.name != str(None):
 665             self.name = ""
 666         else:
 667             self.name = str(self.name)
 668         try:
 669             self.command = str(self.configuration['command'])
 670         except (KeyError, TypeError):
 671             self.error("No command specified. Using: '" + self.command + "'")
 672         self.command = self.command.split(' ')
 673         #if self.command[0] not in self.command_whitelist:
 674         #    self.error("Command is not whitelisted.")
 675         #    return False
 676
 677         for arg in self.command[1:]:
 678             if any(st in arg for st in self.bad_substrings):
 679                 self.error("Bad command argument:" + " ".join(self.command[1:]))
 680                 return False
 681         # test command and search for it in /usr/sbin or /sbin when failed
 682         base = self.command[0].split('/')[-1]
 683         if self._get_raw_data() is None:
 684             for prefix in ['/sbin/', '/usr/sbin/']:
 685                 self.command[0] = prefix + base
 686                 if os.path.isfile(self.command[0]):
 687                     break
 688                 #if self._get_raw_data() is not None:
 689                 #    break
 690
 691         if self._get_data() is None or len(self._get_data()) == 0:
 692             return False
 693         return True