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