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