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 NetSocketService(SimpleService):
 436     def __init__(self, configuration=None, name=None):
 437         self.host = "localhost"
 438         self.port = None
 439         self.sock = None
 440         self.request = ""
 441         SimpleService.__init__(self, configuration=configuration, name=name)
 442
 443     def _get_raw_data(self):
 444         """
 445         Get raw data with low-level "socket" module.
 446         :return: str
 447         """
 448         if self.sock is None:
 449             try:
 450                 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 451                 sock.settimeout(self.update_every)
 452                 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 453                 sock.connect((self.host, self.port))
 454             except Exception as e:
 455                 print(e)
 456                 self.sock = None
 457                 return None
 458
 459         if self.request != "".encode():
 460             try:
 461                 sock.send(self.request)
 462             except Exception:
 463                 try:
 464                     sock.shutdown(1)
 465                     sock.close()
 466                 except:
 467                     pass
 468                 self.sock = None
 469                 return None
 470
 471         data = sock.recv(1024)
 472         try:
 473             while True:
 474                 buf = sock.recv(1024)
 475                 if not buf:
 476                     break
 477                 else:
 478                     data += buf
 479         except:
 480             sock.close()
 481             return None
 482
 483         return data.decode()
 484
 485     def _parse_config(self):
 486         """
 487         Parse configuration data
 488         :return: boolean
 489         """
 490         if self.name is not None or self.name != str(None):
 491             self.name = ""
 492         else:
 493             self.name = str(self.name)
 494         try:
 495             self.host = str(self.configuration['host'])
 496         except (KeyError, TypeError):
 497             self.error("No host specified. Using: '" + self.host + "'")
 498         try:
 499             self.port = int(self.configuration['port'])
 500         except (KeyError, TypeError):
 501             self.error("No port specified. Using: '" + str(self.port) + "'")
 502         try:
 503             self.request = str(self.configuration['request'])
 504         except (KeyError, TypeError):
 505             self.error("No request specified. Using: '" + str(self.request) + "'")
 506         self.request = self.request.encode()
 507
 508
 509 class LogService(SimpleService):
 510     def __init__(self, configuration=None, name=None):
 511         self.log_path = ""
 512         self._last_position = 0
 513         # self._log_reader = None
 514         SimpleService.__init__(self, configuration=configuration, name=name)
 515         self.retries = 100000  # basically always retry
 516
 517     def _get_raw_data(self):
 518         """
 519         Get log lines since last poll
 520         :return: list
 521         """
 522         lines = []
 523         try:
 524             if os.path.getsize(self.log_path) < self._last_position:
 525                 self._last_position = 0
 526             elif os.path.getsize(self.log_path) == self._last_position:
 527                 return None
 528             with open(self.log_path, "r") as fp:
 529                 fp.seek(self._last_position)
 530                 for i, line in enumerate(fp):
 531                     lines.append(line)
 532                 self._last_position = fp.tell()
 533         except Exception as e:
 534             self.error(self.__module__, str(e))
 535
 536         if len(lines) != 0:
 537             return lines
 538         return None
 539
 540     def check(self):
 541         """
 542         Parse basic configuration and check if log file exists
 543         :return: boolean
 544         """
 545         if self.name is not None or self.name != str(None):
 546             self.name = ""
 547         else:
 548             self.name = str(self.name)
 549         try:
 550             self.log_path = str(self.configuration['path'])
 551         except (KeyError, TypeError):
 552             self.error("No path to log specified. Using: '" + self.log_path + "'")
 553
 554         if os.access(self.log_path, os.R_OK):
 555             return True
 556         else:
 557             self.error("Cannot access file: '" + self.log_path + "'")
 558             return False
 559
 560     def create(self):
 561         status = SimpleService.create(self)
 562         self._last_position = 0
 563         return status
 564
 565
 566 class ExecutableService(SimpleService):
 567     command_whitelist = ['exim']
 568
 569     def __init__(self, configuration=None, name=None):
 570         self.command = ""
 571         SimpleService.__init__(self, configuration=configuration, name=name)
 572
 573     def _get_raw_data(self):
 574         """
 575         Get raw data from executed command
 576         :return: str
 577         """
 578         try:
 579             p = Popen(self.command, stdout=PIPE, stderr=PIPE)
 580         except Exception as e:
 581             self.error(self.__module__, str(e))
 582             return None
 583         data = []
 584         for line in p.stdout.readlines():
 585             data.append(line)
 586
 587         return data
 588
 589     def check(self):
 590         """
 591         Parse basic configuration, check if command is whitelisted and is returning values
 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.command = str(self.configuration['path'])
 600         # except (KeyError, TypeError):
 601         #     self.error("No command specified. Using: '" + self.command + "'")
 602         self.command = self.command.split(' ')
 603         for i in self.command:
 604             if i.startswith('-') or i in self.command_whitelist:
 605                 pass
 606             else:
 607                 self.error("Wrong command. Probably not on whitelist.")
 608                 return False
 609         if self._get_data() is None or len(self._get_data()) == 0:
 610             return False
 611         return True