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             if self.name == "":
 334                 type_id = self.__module__
 335             else:
 336                 type_id = self.__module__ + "_" + self.name
 337             self.chart(type_id + "." + name, *options)
 338             # check if server has this datapoint
 339             for line in self.definitions[name]['lines']:
 340                 if line[0] in data:
 341                     self.dimension(*line)
 342             idx += 1
 343
 344         self.commit()
 345         return True
 346
 347     def update(self, interval):
 348         """
 349         Update charts
 350         :param interval: int
 351         :return: boolean
 352         """
 353         data = self._get_data()
 354         if data is None:
 355             return False
 356
 357         updated = False
 358         for chart in self.order:
 359             if str(self.name) == "":
 360                 type_id = self.__module__
 361             else:
 362                 type_id = self.__module__ + "_" + self.name
 363             if self.begin(type_id + "." + chart, interval):
 364                 updated = True
 365                 for dim in self.definitions[chart]['lines']:
 366                     try:
 367                         self.set(dim[0], data[dim[0]])
 368                     except KeyError:
 369                         pass
 370                 self.end()
 371
 372         self.commit()
 373
 374         return updated
 375
 376
 377 class UrlService(SimpleService):
 378     def __init__(self, configuration=None, name=None):
 379         self.url = ""
 380         self.user = None
 381         self.password = None
 382         SimpleService.__init__(self, configuration=configuration, name=name)
 383
 384     def __add_auth(self):
 385         passman = urllib2.HTTPPasswordMgrWithDefaultRealm()
 386         passman.add_password(None, self.url, self.user, self.password)
 387         authhandler = urllib2.HTTPBasicAuthHandler(passman)
 388         opener = urllib2.build_opener(authhandler)
 389         urllib2.install_opener(opener)
 390
 391     def _get_raw_data(self):
 392         """
 393         Get raw data from http request
 394         :return: str
 395         """
 396         raw = None
 397         try:
 398             f = urllib2.urlopen(self.url, timeout=self.update_every)
 399         except Exception as e:
 400             msg.error(self.__module__, str(e))
 401             return None
 402
 403         try:
 404             raw = f.read().decode('utf-8')
 405         except Exception as e:
 406             msg.error(self.__module__, str(e))
 407         finally:
 408             f.close()
 409         return raw
 410
 411     def check(self):
 412         """
 413         Format configuration data and try to connect to server
 414         :return: boolean
 415         """
 416         if self.name is None or self.name == str(None):
 417             self.name = 'local'
 418         else:
 419             self.name = str(self.name)
 420         try:
 421             self.url = str(self.configuration['url'])
 422         except (KeyError, TypeError):
 423             pass
 424         try:
 425             self.user = str(self.configuration['user'])
 426         except (KeyError, TypeError):
 427             pass
 428         try:
 429             self.password = str(self.configuration['password'])
 430         except (KeyError, TypeError):
 431             pass
 432
 433         if self.user is not None and self.password is not None:
 434             self.__add_auth()
 435
 436         if self._get_data() is not None:
 437             return True
 438         else:
 439             return False
 440
 441
 442 class NetSocketService(SimpleService):
 443     def __init__(self, configuration=None, name=None):
 444         self.host = "localhost"
 445         self.port = None
 446         self.sock = None
 447         self.request = ""
 448         SimpleService.__init__(self, configuration=configuration, name=name)
 449
 450     def _get_raw_data(self):
 451         """
 452         Get raw data with low-level "socket" module.
 453         :return: str
 454         """
 455         if self.sock is None:
 456             try:
 457                 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 458                 sock.settimeout(self.update_every)
 459                 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 460                 sock.connect((self.host, self.port))
 461             except Exception as e:
 462                 print(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(1024)
 479         try:
 480             while True:
 481                 buf = sock.recv(1024)
 482                 if not buf:
 483                     break
 484                 else:
 485                     data += buf
 486         except:
 487             sock.close()
 488             return None
 489
 490         return data.decode()
 491
 492     def _parse_config(self):
 493         """
 494         Parse configuration data
 495         :return: boolean
 496         """
 497         if self.name is not None or self.name != str(None):
 498             self.name = ""
 499         else:
 500             self.name = str(self.name)
 501         try:
 502             self.host = str(self.configuration['host'])
 503         except (KeyError, TypeError):
 504             self.error("No host specified. Using: '" + self.host + "'")
 505         try:
 506             self.port = int(self.configuration['port'])
 507         except (KeyError, TypeError):
 508             self.error("No port specified. Using: '" + str(self.port) + "'")
 509         try:
 510             self.request = str(self.configuration['request'])
 511         except (KeyError, TypeError):
 512             self.error("No request specified. Using: '" + str(self.request) + "'")
 513         self.request = self.request.encode()
 514
 515
 516 class LogService(SimpleService):
 517     def __init__(self, configuration=None, name=None):
 518         self.log_path = ""
 519         self._last_position = 0
 520         # self._log_reader = None
 521         SimpleService.__init__(self, configuration=configuration, name=name)
 522         self.retries = 100000  # basically always retry
 523
 524     def _get_raw_data(self):
 525         """
 526         Get log lines since last poll
 527         :return: list
 528         """
 529         lines = []
 530         try:
 531             if os.path.getsize(self.log_path) < self._last_position:
 532                 self._last_position = 0
 533             elif os.path.getsize(self.log_path) == self._last_position:
 534                 return None
 535             with open(self.log_path, "r") as fp:
 536                 fp.seek(self._last_position)
 537                 for i, line in enumerate(fp):
 538                     lines.append(line)
 539                 self._last_position = fp.tell()
 540         except Exception as e:
 541             self.error(self.__module__, str(e))
 542
 543         if len(lines) != 0:
 544             return lines
 545         return None
 546
 547     def check(self):
 548         """
 549         Parse basic configuration and check if log file exists
 550         :return: boolean
 551         """
 552         if self.name is not None or self.name != str(None):
 553             self.name = ""
 554         else:
 555             self.name = str(self.name)
 556         try:
 557             self.log_path = str(self.configuration['path'])
 558         except (KeyError, TypeError):
 559             self.error("No path to log specified. Using: '" + self.log_path + "'")
 560
 561         if os.access(self.log_path, os.R_OK):
 562             return True
 563         else:
 564             self.error("Cannot access file: '" + self.log_path + "'")
 565             return False
 566
 567     def create(self):
 568         status = SimpleService.create(self)
 569         self._last_position = 0
 570         return status
 571
 572
 573 class ExecutableService(SimpleService):
 574     command_whitelist = ['exim']
 575
 576     def __init__(self, configuration=None, name=None):
 577         self.command = ""
 578         SimpleService.__init__(self, configuration=configuration, name=name)
 579
 580     def _get_raw_data(self):
 581         """
 582         Get raw data from executed command
 583         :return: str
 584         """
 585         try:
 586             p = Popen(self.command, stdout=PIPE, stderr=PIPE)
 587         except Exception as e:
 588             self.error(self.__module__, str(e))
 589             return None
 590         data = []
 591         for line in p.stdout.readlines():
 592             data.append(line)
 593
 594         return data
 595
 596     def check(self):
 597         """
 598         Parse basic configuration, check if command is whitelisted and is returning values
 599         :return: boolean
 600         """
 601         if self.name is not None or self.name != str(None):
 602             self.name = ""
 603         else:
 604             self.name = str(self.name)
 605         # try:
 606         #     self.command = str(self.configuration['path'])
 607         # except (KeyError, TypeError):
 608         #     self.error("No command specified. Using: '" + self.command + "'")
 609         self.command = self.command.split(' ')
 610         for i in self.command:
 611             if i.startswith('-') or i in self.command_whitelist:
 612                 pass
 613             else:
 614                 self.error("Wrong command. Probably not on whitelist.")
 615                 return False
 616         if self._get_data() is None or len(self._get_data()) == 0:
 617             return False
 618         return True