]> git.ipfire.org Git - collecty.git/blobdiff - src/collecty/plugins/base.py
Allow getting the standard deviation of the plotted values
[collecty.git] / src / collecty / plugins / base.py
index 94b0bc06593a9ba713b6c34e3d5a1c8e0259536d..c1666b6b3474809d9b7ef90d78e95e0257de4f83 100644 (file)
@@ -1,4 +1,4 @@
-#!/usr/bin/python
+#!/usr/bin/python3
 ###############################################################################
 #                                                                             #
 # collecty - A system statistics collection daemon for IPFire                 #
 #                                                                             #
 ###############################################################################
 
-from __future__ import division
-
 import datetime
 import logging
 import math
 import os
+import re
 import rrdtool
+import tempfile
 import threading
 import time
+import unicodedata
 
+from .. import locales
+from .. import util
 from ..constants import *
 from ..i18n import _
 
-_plugins = {}
-
-def get():
-       """
-               Returns a list with all automatically registered plugins.
-       """
-       return _plugins.values()
+DEF_MATCH = re.compile(r"C?DEF:([A-Za-z0-9_]+)=")
 
 class Timer(object):
        def __init__(self, timeout, heartbeat=1):
@@ -72,7 +69,65 @@ class Timer(object):
                return self.elapsed > self.timeout
 
 
-class Plugin(threading.Thread):
+class Environment(object):
+       """
+               Sets the correct environment for rrdtool to create
+               localised graphs and graphs in the correct timezone.
+       """
+       def __init__(self, timezone, locale):
+               # Build the new environment
+               self.new_environment = {
+                       "TZ" : timezone or DEFAULT_TIMEZONE,
+               }
+
+               for k in ("LANG", "LC_ALL"):
+                       self.new_environment[k] = locale or DEFAULT_LOCALE
+
+       def __enter__(self):
+               # Save the current environment
+               self.old_environment = {}
+               for k in self.new_environment:
+                       self.old_environment[k] = os.environ.get(k, None)
+
+               # Apply the new one
+               os.environ.update(self.new_environment)
+
+       def __exit__(self, type, value, traceback):
+               # Roll back to the previous environment
+               for k, v in self.old_environment.items():
+                       if v is None:
+                               try:
+                                       del os.environ[k]
+                               except KeyError:
+                                       pass
+                       else:
+                               os.environ[k] = v
+
+
+class PluginRegistration(type):
+       plugins = {}
+
+       def __init__(plugin, name, bases, dict):
+               type.__init__(plugin, name, bases, dict)
+
+               # The main class from which is inherited is not registered
+               # as a plugin.
+               if name == "Plugin":
+                       return
+
+               if not all((plugin.name, plugin.description)):
+                       raise RuntimeError(_("Plugin is not properly configured: %s") % plugin)
+
+               PluginRegistration.plugins[plugin.name] = plugin
+
+
+def get():
+       """
+               Returns a list with all automatically registered plugins.
+       """
+       return PluginRegistration.plugins.values()
+
+class Plugin(object, metaclass=PluginRegistration):
        # The name of this plugin.
        name = None
 
@@ -86,26 +141,7 @@ class Plugin(threading.Thread):
        # The default interval for all plugins
        interval = 60
 
-       # Automatically register all providers.
-       class __metaclass__(type):
-               def __init__(plugin, name, bases, dict):
-                       type.__init__(plugin, name, bases, dict)
-
-                       # The main class from which is inherited is not registered
-                       # as a plugin.
-                       if name == "Plugin":
-                               return
-
-                       if not all((plugin.name, plugin.description)):
-                               raise RuntimeError(_("Plugin is not properly configured: %s") \
-                                       % plugin)
-
-                       _plugins[plugin.name] = plugin
-
        def __init__(self, collecty, **kwargs):
-               threading.Thread.__init__(self, name=self.description)
-               self.daemon = True
-
                self.collecty = collecty
 
                # Check if this plugin was configured correctly.
@@ -121,11 +157,7 @@ class Plugin(threading.Thread):
                # Run some custom initialization.
                self.init(**kwargs)
 
-               # Keepalive options
-               self.running = True
-               self.timer = Timer(self.interval)
-
-               self.log.info(_("Successfully initialized %s") % self.__class__.__name__)
+               self.log.debug(_("Successfully initialized %s") % self.__class__.__name__)
 
        @property
        def path(self):
@@ -154,6 +186,8 @@ class Plugin(threading.Thread):
                        now = datetime.datetime.utcnow()
                        try:
                                result = o.collect()
+
+                               result = self._format_result(result)
                        except:
                                self.log.warning(_("Unhandled exception in %s.collect()") % o, exc_info=True)
                                continue
@@ -169,32 +203,77 @@ class Plugin(threading.Thread):
                        self.collecty.write_queue.add(o, now, result)
 
                # Returns the time this function took to complete.
-               return (time.time() - time_start)
+               delay = time.time() - time_start
+
+               # Log some warning when a collect method takes too long to return some data
+               if delay >= 60:
+                       self.log.warning(_("A worker thread was stalled for %.4fs") % delay)
+
+       @staticmethod
+       def _format_result(result):
+               if not isinstance(result, tuple) and not isinstance(result, list):
+                       return result
+
+               # Replace all Nones by NaN
+               s = []
+
+               for e in result:
+                       if e is None:
+                               e = "NaN"
+
+                       # Format as string
+                       e = "%s" % e
+
+                       s.append(e)
+
+               return ":".join(s)
+
+       def get_object(self, id):
+               for object in self.objects:
+                       if not object.id == id:
+                               continue
+
+                       return object
 
-       def run(self):
-               self.log.debug(_("%s plugin has started") % self.name)
+       def get_template(self, template_name, object_id, locale=None, timezone=None):
+               for template in self.templates:
+                       if not template.name == template_name:
+                               continue
 
-               # Initially collect everything
-               self.collect()
+                       return template(self, object_id, locale=locale, timezone=timezone)
 
-               while self.running:
-                       # Reset the timer.
-                       self.timer.reset()
+       def generate_graph(self, template_name, object_id="default",
+                       timezone=None, locale=None, **kwargs):
+               template = self.get_template(template_name, object_id=object_id,
+                       timezone=timezone, locale=locale)
+               if not template:
+                       raise RuntimeError("Could not find template %s" % template_name)
+
+               time_start = time.time()
 
-                       # Wait until the timer has successfully elapsed.
-                       if self.timer.wait():
-                               delay = self.collect()
-                               self.timer.reset(delay)
+               graph = template.generate_graph(**kwargs)
 
-               self.log.debug(_("%s plugin has stopped") % self.name)
+               duration = time.time() - time_start
+               self.log.debug(_("Generated graph %s in %.1fms") \
+                       % (template, duration * 1000))
 
-       def shutdown(self):
-               self.log.debug(_("Received shutdown signal."))
-               self.running = False
+               return graph
 
-               # Kill any running timers.
-               if self.timer:
-                       self.timer.cancel()
+       def graph_info(self, template_name, object_id="default",
+                       timezone=None, locale=None, **kwargs):
+               template = self.get_template(template_name, object_id=object_id,
+                       timezone=timezone, locale=locale)
+               if not template:
+                       raise RuntimeError("Could not find template %s" % template_name)
+
+               return template.graph_info()
+
+       def last_update(self, object_id="default"):
+               object = self.get_object(object_id)
+               if not object:
+                       raise RuntimeError("Could not find object %s" % object_id)
+
+               return object.last_update()
 
 
 class Object(object):
@@ -202,9 +281,12 @@ class Object(object):
        rrd_schema = None
 
        # RRA properties.
-       rra_types     = ["AVERAGE", "MIN", "MAX"]
-       rra_timespans = [3600, 86400, 604800, 2678400, 31622400]
-       rra_rows      = 2880
+       rra_types     = ("AVERAGE", "MIN", "MAX")
+       rra_timespans = (
+               ("1m", "10d"),
+               ("1h", "18M"),
+               ("1d",  "5y"),
+       )
 
        def __init__(self, plugin, *args, **kwargs):
                self.plugin = plugin
@@ -242,7 +324,19 @@ class Object(object):
                """
                        The absolute path to the RRD file of this plugin.
                """
-               return os.path.join(DATABASE_DIR, self.plugin.path, "%s.rrd" % self.id)
+               filename = self._normalise_filename("%s.rrd" % self.id)
+
+               return os.path.join(DATABASE_DIR, self.plugin.path, filename)
+
+       @staticmethod
+       def _normalise_filename(filename):
+               # Convert the filename into ASCII characters only
+               filename = unicodedata.normalize("NFKC", filename)
+
+               # Replace any spaces by dashes
+               filename = filename.replace(" ", "-")
+
+               return filename
 
        ### Basic methods
 
@@ -276,6 +370,39 @@ class Object(object):
        def info(self):
                return rrdtool.info(self.file)
 
+       def last_update(self):
+               """
+                       Returns a dictionary with the timestamp and
+                       data set of the last database update.
+               """
+               return {
+                       "dataset"   : self.last_dataset,
+                       "timestamp" : self.last_updated,
+               }
+
+       def _last_update(self):
+               return rrdtool.lastupdate(self.file)
+
+       @property
+       def last_updated(self):
+               """
+                       Returns the timestamp when this database was last updated
+               """
+               lu = self._last_update()
+
+               if lu:
+                       return lu.get("date")
+
+       @property
+       def last_dataset(self):
+               """
+                       Returns the latest dataset in the database
+               """
+               lu = self._last_update()
+
+               if lu:
+                       return lu.get("ds")
+
        @property
        def stepsize(self):
                return self.plugin.interval
@@ -308,23 +435,51 @@ class Object(object):
 
                xff = 0.1
 
-               cdp_length = 0
-               for rra_timespan in self.rra_timespans:
-                       if (rra_timespan / self.stepsize) < self.rra_rows:
-                               rra_timespan = self.stepsize * self.rra_rows
+               for steps, rows in self.rra_timespans:
+                       for type in self.rra_types:
+                               schema.append("RRA:%s:%s:%s:%s" % (type, xff, steps, rows))
+
+               return schema
+
+       @property
+       def rrd_schema_names(self):
+               ret = []
+
+               for line in self.rrd_schema:
+                       (prefix, name, type, lower_limit, upper_limit) = line.split(":")
+                       ret.append(name)
+
+               return ret
+
+       def make_rrd_defs(self, prefix=None):
+               defs = []
 
-                       if cdp_length == 0:
-                               cdp_length = 1
+               for name in self.rrd_schema_names:
+                       if prefix:
+                               p = "%s_%s" % (prefix, name)
                        else:
-                               cdp_length = rra_timespan // (self.rra_rows * self.stepsize)
+                               p = name
 
-                       cdp_number = math.ceil(rra_timespan / (cdp_length * self.stepsize))
+                       defs += [
+                               "DEF:%s=%s:%s:AVERAGE" % (p, self.file, name),
+                       ]
 
-                       for rra_type in self.rra_types:
-                               schema.append("RRA:%s:%.10f:%d:%d" % \
-                                       (rra_type, xff, cdp_length, cdp_number))
+               return defs
 
-               return schema
+       def get_stddev(self, interval=None):
+               args = self.make_rrd_defs()
+
+               # Add the correct interval
+               args += ["--start", util.make_interval(interval)]
+
+               for name in self.rrd_schema_names:
+                       args += [
+                               "VDEF:%s_stddev=%s,STDEV" % (name, name),
+                               "PRINT:%s_stddev:%%lf" % name,
+                       ]
+
+               x, y, vals = rrdtool.graph("/dev/null", *args)
+               return dict(zip(self.rrd_schema_names, vals))
 
        def execute(self):
                if self.collected:
@@ -343,52 +498,187 @@ class Object(object):
                # Make sure that the RRD database has been created
                self.create()
 
+               # Write everything to disk that is in the write queue
+               self.collecty.write_queue.commit_file(self.file)
+
 
 class GraphTemplate(object):
        # A unique name to identify this graph template.
        name = None
 
+       # Headline of the graph image
+       graph_title = None
+
+       # Vertical label of the graph
+       graph_vertical_label = None
+
+       # Limits
+       lower_limit = None
+       upper_limit = None
+
        # Instructions how to create the graph.
        rrd_graph = None
 
        # Extra arguments passed to rrdgraph.
        rrd_graph_args = []
 
-       def __init__(self, ds):
-               self.ds = ds
+       # Default dimensions for this graph
+       height = GRAPH_DEFAULT_HEIGHT
+       width  = GRAPH_DEFAULT_WIDTH
+
+       def __init__(self, plugin, object_id, locale=None, timezone=None):
+               self.plugin = plugin
+
+               # Save localisation parameters
+               self.locale = locales.get(locale)
+               self.timezone = timezone
+
+               # Get all required RRD objects
+               self.object_id = object_id
+
+               # Get the main object
+               self.object = self.get_object(self.object_id)
+
+       def __repr__(self):
+               return "<%s>" % self.__class__.__name__
 
        @property
        def collecty(self):
-               return self.ds.collecty
+               return self.plugin.collecty
 
-       def graph(self, file, interval=None,
-                       width=GRAPH_DEFAULT_WIDTH, height=GRAPH_DEFAULT_HEIGHT):
-               args = [
-                       "--width", "%d" % width,
-                       "--height", "%d" % height,
+       @property
+       def log(self):
+               return self.plugin.log
+
+       def _make_command_line(self, interval, format=DEFAULT_IMAGE_FORMAT,
+                       width=None, height=None, with_title=True, thumbnail=False):
+               args = [e for e in GRAPH_DEFAULT_ARGUMENTS]
+
+               # Set the default dimensions
+               default_height, default_width = GRAPH_DEFAULT_HEIGHT, GRAPH_DEFAULT_WIDTH
+
+               # A thumbnail doesn't have a legend and other labels
+               if thumbnail:
+                       args.append("--only-graph")
+
+                       default_height = THUMBNAIL_DEFAULT_HEIGHT
+                       default_width = THUMBNAIL_DEFAULT_WIDTH
+
+               args += [
+                       "--imgformat", format,
+                       "--height", "%s" % (height or default_height),
+                       "--width", "%s" % (width or default_width),
                ]
-               args += self.collecty.graph_default_arguments
+
                args += self.rrd_graph_args
 
-               intervals = {
-                       None   : "-3h",
-                       "hour" : "-1h",
-                       "day"  : "-25h",
-                       "week" : "-360h",
-                       "year" : "-365d",
+               # Graph title
+               if with_title and self.graph_title:
+                       args += ["--title", self.graph_title]
+
+               # Vertical label
+               if self.graph_vertical_label:
+                       args += ["--vertical-label", self.graph_vertical_label]
+
+               if self.lower_limit is not None or self.upper_limit is not None:
+                       # Force to honour the set limits
+                       args.append("--rigid")
+
+                       if self.lower_limit is not None:
+                               args += ["--lower-limit", self.lower_limit]
+
+                       if self.upper_limit is not None:
+                               args += ["--upper-limit", self.upper_limit]
+
+               # Add interval
+               args += ["--start", util.make_interval(interval)]
+
+               return args
+
+       def _add_vdefs(self, args):
+               ret = []
+
+               for arg in args:
+                       ret.append(arg)
+
+                       # Search for all DEFs and CDEFs
+                       m = re.match(DEF_MATCH, "%s" % arg)
+                       if m:
+                               name = m.group(1)
+
+                               # Add the VDEFs for minimum, maximum, etc. values
+                               ret += [
+                                       "VDEF:%s_cur=%s,LAST" % (name, name),
+                                       "VDEF:%s_avg=%s,AVERAGE" % (name, name),
+                                       "VDEF:%s_max=%s,MAXIMUM" % (name, name),
+                                       "VDEF:%s_min=%s,MINIMUM" % (name, name),
+                               ]
+
+               return ret
+
+       def get_object(self, *args, **kwargs):
+               return self.plugin.get_object(*args, **kwargs)
+
+       def get_object_table(self):
+               return {
+                       "file" : self.object,
                }
 
-               args.append("--start")
-               try:
-                       args.append(intervals[interval])
-               except KeyError:
-                       args.append(interval)
+       @property
+       def object_table(self):
+               if not hasattr(self, "_object_table"):
+                       self._object_table = self.get_object_table()
 
-               info = { "file" : self.ds.file }
-               for item in self.rrd_graph:
-                       try:
-                               args.append(item % info)
-                       except TypeError:
-                               args.append(item)
+               return self._object_table
+
+       def get_object_files(self):
+               files = {}
+
+               for id, obj in self.object_table.items():
+                       files[id] = obj.file
 
-               rrdtool.graph(file, *args)
+               return files
+
+       def generate_graph(self, interval=None, **kwargs):
+               # Make sure that all collected data is in the database
+               # to get a recent graph image
+               if self.object:
+                       self.object.commit()
+
+               args = self._make_command_line(interval, **kwargs)
+
+               self.log.info(_("Generating graph %s") % self)
+
+               #object_files = self.get_object_files()
+
+               if self.object:
+                       args += self.object.make_rrd_defs()
+
+               args += self.rrd_graph
+               args = self._add_vdefs(args)
+
+               # Convert arguments to string
+               args = [str(e) for e in args]
+
+               for arg in args:
+                       self.log.debug("  %s" % arg)
+
+               with Environment(self.timezone, self.locale.lang):
+                       graph = rrdtool.graphv("-", *args)
+
+               return {
+                       "image"        : graph.get("image"),
+                       "image_height" : graph.get("image_height"),
+                       "image_width"  : graph.get("image_width"),
+               }
+
+       def graph_info(self):
+               """
+                       Returns a dictionary with useful information
+                       about this graph.
+               """
+               return {
+                       "title"        : self.graph_title,
+                       "object_id"    : self.object_id or "",
+                       "template"     : self.name,
+               }