collecty/plugins/base.py

   1 #!/usr/bin/python
   2 ###############################################################################
   3 #                                                                             #
   4 # collecty - A system statistics collection daemon for IPFire                 #
   5 # Copyright (C) 2012 IPFire development team                                  #
   6 #                                                                             #
   7 # This program is free software: you can redistribute it and/or modify        #
   8 # it under the terms of the GNU General Public License as published by        #
   9 # the Free Software Foundation, either version 3 of the License, or           #
  10 # (at your option) any later version.                                         #
  11 #                                                                             #
  12 # This program is distributed in the hope that it will be useful,             #
  13 # but WITHOUT ANY WARRANTY; without even the implied warranty of              #
  14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the               #
  15 # GNU General Public License for more details.                                #
  16 #                                                                             #
  17 # You should have received a copy of the GNU General Public License           #
  18 # along with this program.  If not, see <http://www.gnu.org/licenses/>.       #
  19 #                                                                             #
  20 ###############################################################################
  21
  22 from __future__ import division
  23
  24 import logging
  25 import math
  26 import os
  27 import rrdtool
  28 import threading
  29 import time
  30
  31 from ..constants import *
  32 from ..i18n import _
  33
  34 class Timer(object):
  35         def __init__(self, timeout, heartbeat=1):
  36                 self.timeout = timeout
  37                 self.heartbeat = heartbeat
  38
  39                 self.delay = 0
  40
  41                 self.reset()
  42
  43         def reset(self, delay=0):
  44                 # Save start time.
  45                 self.start = time.time()
  46
  47                 self.delay = delay
  48
  49                 # Has this timer been killed?
  50                 self.killed = False
  51
  52         @property
  53         def elapsed(self):
  54                 return time.time() - self.start - self.delay
  55
  56         def cancel(self):
  57                 self.killed = True
  58
  59         def wait(self):
  60                 while self.elapsed < self.timeout and not self.killed:
  61                         time.sleep(self.heartbeat)
  62
  63                 return self.elapsed > self.timeout
  64
  65
  66 class DataSource(threading.Thread):
  67         # The name of this plugin.
  68         name = None
  69
  70         # A description for this plugin.
  71         description = None
  72
  73         # Templates which can be used to generate a graph out of
  74         # the data from this data source.
  75         templates = []
  76
  77         # The schema of the RRD database.
  78         rrd_schema = None
  79
  80         # RRA properties.
  81         rra_types = ["AVERAGE", "MIN", "MAX"]
  82         rra_timespans = [3600, 86400, 604800, 2678400, 31622400]
  83         rra_rows = 2880
  84
  85         # The default interval of this plugin.
  86         default_interval = 60
  87
  88         def __init__(self, collecty, **kwargs):
  89                 threading.Thread.__init__(self, name=self.description)
  90                 self.daemon = True
  91
  92                 self.collecty = collecty
  93
  94                 # Check if this plugin was configured correctly.
  95                 assert self.name, "Name of the plugin is not set: %s" % self.name
  96                 assert self.description, "Description of the plugin is not set: %s" % self.description
  97                 assert self.rrd_schema
  98
  99                 # Initialize the logger.
 100                 self.log = logging.getLogger("collecty.plugins.%s" % self.name)
 101                 self.log.propagate = 1
 102
 103                 self.data = []
 104
 105                 # Run some custom initialization.
 106                 self.init(**kwargs)
 107
 108                 # Create the database file.
 109                 self.create()
 110
 111                 # Keepalive options
 112                 self.running = True
 113                 self.timer = Timer(self.interval)
 114
 115                 self.log.info(_("Successfully initialized (%s).") % self.id)
 116
 117         def __repr__(self):
 118                 return "<%s %s>" % (self.__class__.__name__, self.id)
 119
 120         @property
 121         def id(self):
 122                 """
 123                         A unique ID of the plugin instance.
 124                 """
 125                 return self.name
 126
 127         @property
 128         def interval(self):
 129                 """
 130                         Returns the interval in milliseconds, when the read method
 131                         should be called again.
 132                 """
 133                 # XXX read this from the settings
 134
 135                 # Otherwise return the default.
 136                 return self.default_interval
 137
 138         @property
 139         def stepsize(self):
 140                 return self.interval
 141
 142         @property
 143         def file(self):
 144                 """
 145                         The absolute path to the RRD file of this plugin.
 146                 """
 147                 return os.path.join(DATABASE_DIR, "%s.rrd" % self.id)
 148
 149         def create(self):
 150                 """
 151                         Creates an empty RRD file with the desired data structures.
 152                 """
 153                 # Skip if the file does already exist.
 154                 if os.path.exists(self.file):
 155                         return
 156
 157                 dirname = os.path.dirname(self.file)
 158                 if not os.path.exists(dirname):
 159                         os.makedirs(dirname)
 160
 161                 # Create argument list.
 162                 args = [
 163                         "--step", "%s" % self.default_interval,
 164                 ] + self.get_rrd_schema()
 165
 166                 rrdtool.create(self.file, *args)
 167
 168                 self.log.debug(_("Created RRD file %s.") % self.file)
 169
 170         def get_rrd_schema(self):
 171                 schema = [
 172                         "--step", "%s" % self.stepsize,
 173                 ]
 174                 for line in self.rrd_schema:
 175                         if line.startswith("DS:"):
 176                                 try:
 177                                         (prefix, name, type, lower_limit, upper_limit) = line.split(":")
 178
 179                                         line = ":".join((
 180                                                 prefix,
 181                                                 name,
 182                                                 type,
 183                                                 "%s" % self.stepsize,
 184                                                 lower_limit,
 185                                                 upper_limit
 186                                         ))
 187                                 except ValueError:
 188                                         pass
 189
 190                         schema.append(line)
 191
 192                 xff = 0.1
 193
 194                 cdp_length = 0
 195                 for rra_timespan in self.rra_timespans:
 196                         if (rra_timespan / self.stepsize) < self.rra_rows:
 197                                 rra_timespan = self.stepsize * self.rra_rows
 198
 199                         if cdp_length == 0:
 200                                 cdp_length = 1
 201                         else:
 202                                 cdp_length = rra_timespan // (self.rra_rows * self.stepsize)
 203
 204                         cdp_number = math.ceil(rra_timespan / (cdp_length * self.stepsize))
 205
 206                         for rra_type in self.rra_types:
 207                                 schema.append("RRA:%s:%.10f:%d:%d" % \
 208                                         (rra_type, xff, cdp_length, cdp_number))
 209
 210                 return schema
 211
 212         def info(self):
 213                 return rrdtool.info(self.file)
 214
 215         ### Basic methods
 216
 217         def init(self, **kwargs):
 218                 """
 219                         Do some custom initialization stuff here.
 220                 """
 221                 pass
 222
 223         def read(self):
 224                 """
 225                         Gathers the statistical data, this plugin collects.
 226                 """
 227                 raise NotImplementedError
 228
 229         def submit(self):
 230                 """
 231                         Flushes the read data to disk.
 232                 """
 233                 # Do nothing in case there is no data to submit.
 234                 if not self.data:
 235                         return
 236
 237                 self.log.debug(_("Submitting data to database. %d entries.") % len(self.data))
 238                 rrdtool.update(self.file, *self.data)
 239                 self.data = []
 240
 241         def _read(self, *args, **kwargs):
 242                 """
 243                         This method catches errors from the read() method and logs them.
 244                 """
 245                 try:
 246                         return self.read(*args, **kwargs)
 247
 248                 # Catch any exceptions, so collecty does not crash.
 249                 except Exception, e:
 250                         self.log.critical(_("Unhandled exception in read()!"), exc_info=True)
 251
 252         def _submit(self, *args, **kwargs):
 253                 """
 254                         This method catches errors from the submit() method and logs them.
 255                 """
 256                 try:
 257                         return self.submit(*args, **kwargs)
 258
 259                 # Catch any exceptions, so collecty does not crash.
 260                 except Exception, e:
 261                         self.log.critical(_("Unhandled exception in submit()!"), exc_info=True)
 262
 263         def run(self):
 264                 self.log.debug(_("Started."))
 265
 266                 while self.running:
 267                         # Reset the timer.
 268                         self.timer.reset()
 269
 270                         # Wait until the timer has successfully elapsed.
 271                         if self.timer.wait():
 272                                 self.log.debug(_("Collecting..."))
 273                                 self._read()
 274
 275                 self._submit()
 276                 self.log.debug(_("Stopped."))
 277
 278         def shutdown(self):
 279                 self.log.debug(_("Received shutdown signal."))
 280                 self.running = False
 281
 282                 # Kill any running timers.
 283                 if self.timer:
 284                         self.timer.cancel()
 285
 286         @property
 287         def now(self):
 288                 """
 289                         Returns the current timestamp in the UNIX timestamp format (UTC).
 290                 """
 291                 return int(time.time())
 292
 293
 294 class GraphTemplate(object):
 295         # A unique name to identify this graph template.
 296         name = None
 297
 298         # Instructions how to create the graph.
 299         rrd_graph = None
 300
 301         # Extra arguments passed to rrdgraph.
 302         rrd_graph_args = []
 303
 304         def __init__(self, ds):
 305                 self.ds = ds
 306
 307         @property
 308         def collecty(self):
 309                 return self.ds.collecty
 310
 311         def graph(self, file, interval=None,
 312                         width=GRAPH_DEFAULT_WIDTH, height=GRAPH_DEFAULT_HEIGHT):
 313                 args = [
 314                         "--width", "%d" % width,
 315                         "--height", "%d" % height,
 316                 ]
 317                 args += self.collecty.graph_default_arguments
 318                 args += self.rrd_graph_args
 319
 320                 intervals = {
 321                         None   : "-3h",
 322                         "hour" : "-1h",
 323                         "day"  : "-25h",
 324                         "week" : "-360h",
 325                         "year" : "-365d",
 326                 }
 327
 328                 args.append("--start")
 329                 try:
 330                         args.append(intervals[interval])
 331                 except KeyError:
 332                         args.append(interval)
 333
 334                 info = { "file" : self.ds.file }
 335                 for item in self.rrd_graph:
 336                         try:
 337                                 args.append(item % info)
 338                         except TypeError:
 339                                 args.append(item)
 340
 341                 rrdtool.graph(file, *args)