src/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 heartbeat(self):
 144                 return self.stepsize * 2
 145
 146         @property
 147         def file(self):
 148                 """
 149                         The absolute path to the RRD file of this plugin.
 150                 """
 151                 return os.path.join(DATABASE_DIR, "%s.rrd" % self.id)
 152
 153         def create(self):
 154                 """
 155                         Creates an empty RRD file with the desired data structures.
 156                 """
 157                 # Skip if the file does already exist.
 158                 if os.path.exists(self.file):
 159                         return
 160
 161                 dirname = os.path.dirname(self.file)
 162                 if not os.path.exists(dirname):
 163                         os.makedirs(dirname)
 164
 165                 # Create argument list.
 166                 args = self.get_rrd_schema()
 167
 168                 rrdtool.create(self.file, *args)
 169
 170                 self.log.debug(_("Created RRD file %s.") % self.file)
 171                 for arg in args:
 172                         self.log.debug("  %s" % arg)
 173
 174         def get_rrd_schema(self):
 175                 schema = [
 176                         "--step", "%s" % self.stepsize,
 177                 ]
 178                 for line in self.rrd_schema:
 179                         if line.startswith("DS:"):
 180                                 try:
 181                                         (prefix, name, type, lower_limit, upper_limit) = line.split(":")
 182
 183                                         line = ":".join((
 184                                                 prefix,
 185                                                 name,
 186                                                 type,
 187                                                 "%s" % self.heartbeat,
 188                                                 lower_limit,
 189                                                 upper_limit
 190                                         ))
 191                                 except ValueError:
 192                                         pass
 193
 194                         schema.append(line)
 195
 196                 xff = 0.1
 197
 198                 cdp_length = 0
 199                 for rra_timespan in self.rra_timespans:
 200                         if (rra_timespan / self.stepsize) < self.rra_rows:
 201                                 rra_timespan = self.stepsize * self.rra_rows
 202
 203                         if cdp_length == 0:
 204                                 cdp_length = 1
 205                         else:
 206                                 cdp_length = rra_timespan // (self.rra_rows * self.stepsize)
 207
 208                         cdp_number = math.ceil(rra_timespan / (cdp_length * self.stepsize))
 209
 210                         for rra_type in self.rra_types:
 211                                 schema.append("RRA:%s:%.10f:%d:%d" % \
 212                                         (rra_type, xff, cdp_length, cdp_number))
 213
 214                 return schema
 215
 216         def info(self):
 217                 return rrdtool.info(self.file)
 218
 219         ### Basic methods
 220
 221         def init(self, **kwargs):
 222                 """
 223                         Do some custom initialization stuff here.
 224                 """
 225                 pass
 226
 227         def read(self):
 228                 """
 229                         Gathers the statistical data, this plugin collects.
 230                 """
 231                 raise NotImplementedError
 232
 233         def submit(self):
 234                 """
 235                         Flushes the read data to disk.
 236                 """
 237                 # Do nothing in case there is no data to submit.
 238                 if not self.data:
 239                         return
 240
 241                 self.log.debug(_("Submitting data to database. %d entries.") % len(self.data))
 242                 for data in self.data:
 243                         self.log.debug("  %s" % data)
 244
 245                 # Create the RRD files (if they don't exist yet or
 246                 # have vanished for some reason).
 247                 self.create()
 248
 249                 rrdtool.update(self.file, *self.data)
 250                 self.data = []
 251
 252         def _read(self, *args, **kwargs):
 253                 """
 254                         This method catches errors from the read() method and logs them.
 255                 """
 256                 start_time = time.time()
 257
 258                 try:
 259                         data = self.read(*args, **kwargs)
 260                         if data is None:
 261                                 self.log.warning(_("Received empty data."))
 262                         else:
 263                                 self.data.append("%d:%s" % (start_time, data))
 264
 265                 # Catch any exceptions, so collecty does not crash.
 266                 except Exception, e:
 267                         self.log.critical(_("Unhandled exception in read()!"), exc_info=True)
 268
 269                 # Return the elapsed time since _read() has been called.
 270                 return (time.time() - start_time)
 271
 272         def _submit(self, *args, **kwargs):
 273                 """
 274                         This method catches errors from the submit() method and logs them.
 275                 """
 276                 try:
 277                         return self.submit(*args, **kwargs)
 278
 279                 # Catch any exceptions, so collecty does not crash.
 280                 except Exception, e:
 281                         self.log.critical(_("Unhandled exception in submit()!"), exc_info=True)
 282
 283         def run(self):
 284                 self.log.debug(_("Started."))
 285
 286                 while self.running:
 287                         # Reset the timer.
 288                         self.timer.reset()
 289
 290                         # Wait until the timer has successfully elapsed.
 291                         if self.timer.wait():
 292                                 self.log.debug(_("Collecting..."))
 293                                 delay = self._read()
 294
 295                                 self.timer.reset(delay)
 296
 297                 self._submit()
 298                 self.log.debug(_("Stopped."))
 299
 300         def shutdown(self):
 301                 self.log.debug(_("Received shutdown signal."))
 302                 self.running = False
 303
 304                 # Kill any running timers.
 305                 if self.timer:
 306                         self.timer.cancel()
 307
 308
 309 class GraphTemplate(object):
 310         # A unique name to identify this graph template.
 311         name = None
 312
 313         # Instructions how to create the graph.
 314         rrd_graph = None
 315
 316         # Extra arguments passed to rrdgraph.
 317         rrd_graph_args = []
 318
 319         def __init__(self, ds):
 320                 self.ds = ds
 321
 322         @property
 323         def collecty(self):
 324                 return self.ds.collecty
 325
 326         def graph(self, file, interval=None,
 327                         width=GRAPH_DEFAULT_WIDTH, height=GRAPH_DEFAULT_HEIGHT):
 328                 args = [
 329                         "--width", "%d" % width,
 330                         "--height", "%d" % height,
 331                 ]
 332                 args += self.collecty.graph_default_arguments
 333                 args += self.rrd_graph_args
 334
 335                 intervals = {
 336                         None   : "-3h",
 337                         "hour" : "-1h",
 338                         "day"  : "-25h",
 339                         "week" : "-360h",
 340                         "year" : "-365d",
 341                 }
 342
 343                 args.append("--start")
 344                 try:
 345                         args.append(intervals[interval])
 346                 except KeyError:
 347                         args.append(interval)
 348
 349                 info = { "file" : self.ds.file }
 350                 for item in self.rrd_graph:
 351                         try:
 352                                 args.append(item % info)
 353                         except TypeError:
 354                                 args.append(item)
 355
 356                 rrdtool.graph(file, *args)