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     # from urllib.request import urlopen, Request, HTTPPasswordMgrWithDefaultRealm
  11     import urllib.request as urllib2
  12 except ImportError:
  13     # from urllib2 import urlopen, Request, HTTPPasswordMgrWithDefaultRealm
  14     import urllib2
  15
  16 # from subprocess import STDOUT, PIPE, Popen
  17 import threading
  18 import msg
  19
  20
  21 class BaseService(threading.Thread):
  22     """
  23     Prototype of Service class.
  24     Implemented basic functionality to run jobs by `python.d.plugin`
  25     """
  26     def __init__(self, configuration=None, name=None):
  27         """
  28         This needs to be initialized in child classes
  29         :param configuration: dict
  30         :param name: str
  31         """
  32         threading.Thread.__init__(self)
  33         self._data_stream = ""
  34         self.daemon = True
  35         self.retries = 0
  36         self.retries_left = 0
  37         self.priority = 140000
  38         self.update_every = 1
  39         self.name = name
  40         self.override_name = None
  41         self.chart_name = ""
  42         self._dimensions = []
  43         self._charts = []
  44         if configuration is None:
  45             self.error("BaseService: no configuration parameters supplied. Cannot create Service.")
  46             raise RuntimeError
  47         else:
  48             self._extract_base_config(configuration)
  49             self.timetable = {}
  50             self.create_timetable()
  51
  52     def _extract_base_config(self, config):
  53         """
  54         Get basic parameters to run service
  55         Minimum config:
  56             config = {'update_every':1,
  57                       'priority':100000,
  58                       'retries':0}
  59         :param config: dict
  60         """
  61         try:
  62             self.override_name = config.pop('override_name')
  63         except KeyError:
  64             pass
  65         self.update_every = int(config.pop('update_every'))
  66         self.priority = int(config.pop('priority'))
  67         self.retries = int(config.pop('retries'))
  68         self.retries_left = self.retries
  69         self.configuration = config
  70
  71     def create_timetable(self, freq=None):
  72         """
  73         Create service timetable.
  74         `freq` is optional
  75         Example:
  76             timetable = {'last': 1466370091.3767564,
  77                          'next': 1466370092,
  78                          'freq': 1}
  79         :param freq: int
  80         """
  81         if freq is None:
  82             freq = self.update_every
  83         now = time.time()
  84         self.timetable = {'last': now,
  85                           'next': now - (now % freq) + freq,
  86                           'freq': freq}
  87
  88     def _run_once(self):
  89         """
  90         Executes self.update(interval) and draws run time chart.
  91         Return value presents exit status of update()
  92         :return: boolean
  93         """
  94         t_start = time.time()
  95         # check if it is time to execute job update() function
  96         if self.timetable['next'] > t_start:
  97             msg.debug(self.chart_name + " will be run in " +
  98                       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         if not self.update(since_last):
 107             return False
 108         t_end = time.time()
 109         self.timetable['next'] = t_end - (t_end % self.timetable['freq']) + self.timetable['freq']
 110
 111         # draw performance graph
 112         run_time = str(int((t_end - t_start) * 1000))
 113         run_time_chart = "BEGIN netdata.plugin_pythond_" + self.chart_name + " " + str(since_last) + '\n'
 114         run_time_chart += "SET run_time = " + run_time + '\n'
 115         run_time_chart += "END\n"
 116         sys.stdout.write(run_time_chart)
 117         msg.debug(self.chart_name + " updated in " + str(run_time) + " ms")
 118         self.timetable['last'] = t_start
 119         return True
 120
 121     def run(self):
 122         """
 123         Runs job in thread. Handles retries.
 124         Exits when job failed or timed out.
 125         :return: None
 126         """
 127         self.timetable['last'] = time.time()
 128         while True:
 129             try:
 130                 status = self._run_once()
 131             except Exception as e:
 132                 msg.error("Something wrong: " + str(e))
 133                 return
 134             if status:
 135                 time.sleep(self.timetable['next'] - time.time())
 136                 self.retries_left = self.retries
 137             else:
 138                 self.retries_left -= 1
 139                 if self.retries_left <= 0:
 140                     msg.error("no more retries. Exiting")
 141                     return
 142                 else:
 143                     time.sleep(self.timetable['freq'])
 144
 145     def _line(self, instruction, *params):
 146         """
 147         Converts *params to string and joins them with one space between every one.
 148         :param params: str/int/float
 149         """
 150         self._data_stream += instruction
 151         for p in params:
 152             if p is None:
 153                 p = ""
 154             else:
 155                 p = str(p)
 156             if len(p) == 0:
 157                 p = "''"
 158             if ' ' in p:
 159                 p = "'" + p + "'"
 160             self._data_stream += " " + p
 161         self._data_stream += "\n"
 162
 163     def chart(self, type_id, name="", title="", units="", family="",
 164               category="", charttype="line", priority="", update_every=""):
 165         """
 166         Defines a new chart.
 167         :param type_id: str
 168         :param name: str
 169         :param title: str
 170         :param units: str
 171         :param family: str
 172         :param category: str
 173         :param charttype: str
 174         :param priority: int/str
 175         :param update_every: int/str
 176         """
 177         self._charts.append(type_id)
 178         self._line("CHART", type_id, name, title, units, family, category, charttype, priority, update_every)
 179
 180     def dimension(self, id, name=None, algorithm="absolute", multiplier=1, divisor=1, hidden=False):
 181         """
 182         Defines a new dimension for the chart
 183         :param id: str
 184         :param name: str
 185         :param algorithm: str
 186         :param multiplier: int/str
 187         :param divisor: int/str
 188         :param hidden: boolean
 189         :return:
 190         """
 191         try:
 192             int(multiplier)
 193         except TypeError:
 194             self.error("malformed dimension: multiplier is not a number:", multiplier)
 195             multiplier = 1
 196         try:
 197             int(divisor)
 198         except TypeError:
 199             self.error("malformed dimension: divisor is not a number:", divisor)
 200             divisor = 1
 201         if name is None:
 202             name = id
 203         if algorithm not in ("absolute", "incremental", "percentage-of-absolute-row", "percentage-of-incremental-row"):
 204             algorithm = "absolute"
 205
 206         self._dimensions.append(id)
 207         if hidden:
 208             self._line("DIMENSION", id, name, algorithm, multiplier, divisor, "hidden")
 209         else:
 210             self._line("DIMENSION", id, name, algorithm, multiplier, divisor)
 211
 212     def begin(self, type_id, microseconds=0):
 213         """
 214         Begin data set
 215         :param type_id: str
 216         :param microseconds: int
 217         :return: boolean
 218         """
 219         if type_id not in self._charts:
 220             self.error("wrong chart type_id:", type_id)
 221             return False
 222         try:
 223             int(microseconds)
 224         except TypeError:
 225             self.error("malformed begin statement: microseconds are not a number:", microseconds)
 226             microseconds = ""
 227
 228         self._line("BEGIN", type_id, microseconds)
 229         return True
 230
 231     def set(self, id, value):
 232         """
 233         Set value to dimension
 234         :param id: str
 235         :param value: int/float
 236         :return: boolean
 237         """
 238         if id not in self._dimensions:
 239             self.error("wrong dimension id:", id)
 240             return False
 241         try:
 242             value = str(int(value))
 243         except TypeError:
 244             self.error("cannot set non-numeric value:", value)
 245             return False
 246         self._line("SET", id, "=", value)
 247         return True
 248
 249     def end(self):
 250         self._line("END")
 251
 252     def commit(self):
 253         """
 254         Upload new data to netdata
 255         """
 256         print(self._data_stream)
 257         self._data_stream = ""
 258
 259     def error(self, *params):
 260         """
 261         Show error message on stderr
 262         """
 263         msg.error(self.chart_name, *params)
 264
 265     def debug(self, *params):
 266         """
 267         Show debug message on stderr
 268         """
 269         msg.debug(self.chart_name, *params)
 270
 271     def info(self, *params):
 272         """
 273         Show information message on stderr
 274         """
 275         msg.info(self.chart_name, *params)
 276
 277     def check(self):
 278         """
 279         check() prototype
 280         :return: boolean
 281         """
 282         msg.error("Service " + str(self.__module__) + "doesn't implement check() function")
 283         return False
 284
 285     def create(self):
 286         """
 287         create() prototype
 288         :return: boolean
 289         """
 290         msg.error("Service " + str(self.__module__) + "doesn't implement create() function?")
 291         return False
 292
 293     def update(self, interval):
 294         """
 295         update() prototype
 296         :param interval: int
 297         :return: boolean
 298         """
 299         msg.error("Service " + str(self.__module__) + "doesn't implement update() function")
 300         return False
 301
 302
 303 class SimpleService(BaseService):
 304     def __init__(self, configuration=None, name=None):
 305         self.order = []
 306         self.definitions = {}
 307         BaseService.__init__(self, configuration=configuration, name=name)
 308
 309     def _get_data(self):
 310         """
 311         Get some data
 312         :return: dict
 313         """
 314         return {}
 315
 316     def check(self):
 317         """
 318         :return:
 319         """
 320         return True
 321
 322     def create(self):
 323         """
 324         Create charts
 325         :return: boolean
 326         """
 327         data = self._get_data()
 328         if data is None:
 329             return False
 330
 331         idx = 0
 332         for name in self.order:
 333             options = self.definitions[name]['options'] + [self.priority + idx, self.update_every]
 334             self.chart(self.__module__ + "_" + self.name + "." + name, *options)
 335             # check if server has this datapoint
 336             for line in self.definitions[name]['lines']:
 337                 if line[0] in data:
 338                     self.dimension(*line)
 339             idx += 1
 340
 341         self.commit()
 342         return True
 343
 344     def update(self, interval):
 345         """
 346         Update charts
 347         :param interval: int
 348         :return: boolean
 349         """
 350         data = self._get_data()
 351         if data is None:
 352             return False
 353
 354         updated = False
 355         for chart in self.order:
 356             if self.begin(self.__module__ + "_" + str(self.name) + "." + chart, interval):
 357                 updated = True
 358                 for dim in self.definitions[chart]['lines']:
 359                     try:
 360                         self.set(dim[0], data[dim[0]])
 361                     except KeyError:
 362                         pass
 363                 self.end()
 364
 365         self.commit()
 366
 367         return updated
 368
 369
 370 class UrlService(SimpleService):
 371     def __init__(self, configuration=None, name=None):
 372         self.url = ""
 373         self.user = None
 374         self.password = None
 375         SimpleService.__init__(self, configuration=configuration, name=name)
 376
 377     def __add_auth(self):
 378         passman = urllib2.HTTPPasswordMgrWithDefaultRealm()
 379         passman.add_password(None, self.url, self.user, self.password)
 380         authhandler = urllib2.HTTPBasicAuthHandler(passman)
 381         opener = urllib2.build_opener(authhandler)
 382         urllib2.install_opener(opener)
 383
 384     def _get_raw_data(self):
 385         """
 386         Get raw data from http request
 387         :return: str
 388         """
 389         raw = None
 390         try:
 391             f = urllib2.urlopen(self.url, timeout=self.update_every)
 392         except Exception as e:
 393             msg.error(self.__module__, str(e))
 394             return None
 395
 396         try:
 397             raw = f.read().decode('utf-8')
 398         except Exception as e:
 399             msg.error(self.__module__, str(e))
 400         finally:
 401             f.close()
 402         return raw
 403
 404     def check(self):
 405         """
 406         Format configuration data and try to connect to server
 407         :return: boolean
 408         """
 409         if self.name is None or self.name == str(None):
 410             self.name = 'local'
 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                 self.sock = None
 456                 return None
 457
 458         try:
 459             sock.send(self.request)
 460         except Exception:
 461             try:
 462                 sock.shutdown(1)
 463                 sock.close()
 464             except:
 465                 pass
 466             self.sock = None
 467             return None
 468
 469         data = sock.recv(1024)
 470         try:
 471             while True:
 472                 buf = sock.recv(1024)
 473                 if not buf:
 474                     break
 475                 else:
 476                     data += buf
 477         except:
 478             sock.close()
 479             return None
 480
 481         return data.decode()
 482
 483     def _parse_config(self):
 484         """
 485         Parse configuration data
 486         :return: boolean
 487         """
 488         if self.name is not None or self.name != str(None):
 489             self.name = ""
 490         else:
 491             self.name = str(self.name)
 492         try:
 493             self.host = str(self.configuration['host'])
 494         except (KeyError, TypeError):
 495             self.error("No host specified. Using: '" + self.host + "'")
 496         try:
 497             self.port = int(self.configuration['port'])
 498         except (KeyError, TypeError):
 499             self.error("No port specified. Using: '" + str(self.port) + "'")
 500         try:
 501             self.port = int(self.configuration['request'])
 502         except (KeyError, TypeError):
 503             self.error("No request specified. Using: '" + str(self.request) + "'")
 504         self.request = self.request.encode()
 505
 506
 507 class LogService(SimpleService):
 508     def __init__(self, configuration=None, name=None):
 509         self.log_path = ""
 510         self._last_position = 0
 511         # self._log_reader = None
 512         SimpleService.__init__(self, configuration=configuration, name=name)
 513         self.retries = 100000  # basically always retry
 514
 515     def _get_raw_data(self):
 516         """
 517         Get log lines since last poll
 518         :return: list
 519         """
 520         lines = []
 521         try:
 522             if os.path.getsize(self.log_path) < self._last_position:
 523                 self._last_position = 0
 524             elif os.path.getsize(self.log_path) == self._last_position:
 525                 return None
 526             with open(self.log_path, "r") as fp:
 527                 fp.seek(self._last_position)
 528                 for i, line in enumerate(fp):
 529                     lines.append(line)
 530                 self._last_position = fp.tell()
 531         except Exception as e:
 532             msg.error(self.__module__, str(e))
 533
 534         if len(lines) != 0:
 535             return lines
 536         return None
 537
 538     def check(self):
 539         """
 540         Parse basic configuration and check if log file exists
 541         :return: boolean
 542         """
 543         if self.name is not None or self.name != str(None):
 544             self.name = ""
 545         else:
 546             self.name = str(self.name)
 547         try:
 548             self.log_path = str(self.configuration['path'])
 549         except (KeyError, TypeError):
 550             self.error("No path to log specified. Using: '" + self.log_path + "'")
 551
 552         if os.access(self.log_path, os.R_OK):
 553             return True
 554         else:
 555             self.error("Cannot access file: '" + self.log_path + "'")
 556             return False
 557
 558     def create(self):
 559         status = SimpleService.create(self)
 560         self._last_position = 0
 561         return status
 562