]> git.ipfire.org Git - collecty.git/blobdiff - src/collecty/plugins/base.py
Refectoring of the main classes
[collecty.git] / src / collecty / plugins / base.py
index 74bc7122a812fe8935d2e36dd6eeda84ec67843c..5b130444324f5c530efdad3b97e3155888d35c4e 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):
@@ -73,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
 
@@ -87,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.
@@ -122,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):
@@ -155,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
@@ -170,32 +203,30 @@ 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
 
-       def run(self):
-               self.log.debug(_("%s plugin has started") % self.name)
+               # 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)
 
-               # Initially collect everything
-               self.collect()
+       @staticmethod
+       def _format_result(result):
+               if not isinstance(result, tuple) and not isinstance(result, list):
+                       return result
 
-               while self.running:
-                       # Reset the timer.
-                       self.timer.reset()
+               # Replace all Nones by NaN
+               s = []
 
-                       # Wait until the timer has successfully elapsed.
-                       if self.timer.wait():
-                               delay = self.collect()
-                               self.timer.reset(delay)
+               for e in result:
+                       if e is None:
+                               e = "NaN"
 
-               self.log.debug(_("%s plugin has stopped") % self.name)
+                       # Format as string
+                       e = "%s" % e
 
-       def shutdown(self):
-               self.log.debug(_("Received shutdown signal."))
-               self.running = False
+                       s.append(e)
 
-               # Kill any running timers.
-               if self.timer:
-                       self.timer.cancel()
+               return ":".join(s)
 
        def get_object(self, id):
                for object in self.objects:
@@ -204,37 +235,58 @@ class Plugin(threading.Thread):
 
                        return object
 
-       def get_template(self, template_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
 
-                       return template(self)
+                       return template(self, object_id, locale=locale, timezone=timezone)
 
-       def generate_graph(self, template_name, object_id="default", **kwargs):
-               template = self.get_template(template_name)
+       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()
 
-               graph = template.generate_graph(object_id=object_id, **kwargs)
+               graph = template.generate_graph(**kwargs)
 
                duration = time.time() - time_start
-               self.log.info(_("Generated graph %s in %.1fms") \
+               self.log.debug(_("Generated graph %s in %.1fms") \
                        % (template, duration * 1000))
 
                return graph
 
+       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):
        # The schema of the RRD database.
        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
@@ -251,6 +303,9 @@ class Object(object):
        def __repr__(self):
                return "<%s>" % self.__class__.__name__
 
+       def __lt__(self, other):
+               return self.id < other.id
+
        @property
        def collecty(self):
                return self.plugin.collecty
@@ -272,7 +327,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
 
@@ -306,6 +373,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
@@ -338,23 +438,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
 
-                       if cdp_length == 0:
-                               cdp_length = 1
+       def make_rrd_defs(self, prefix=None):
+               defs = []
+
+               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:
@@ -373,32 +501,48 @@ 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 = []
 
-       intervals = {
-               None   : "-3h",
-               "hour" : "-1h",
-               "day"  : "-25h",
-               "week" : "-360h",
-               "year" : "-365d",
-       }
-
        # Default dimensions for this graph
        height = GRAPH_DEFAULT_HEIGHT
        width  = GRAPH_DEFAULT_WIDTH
 
-       def __init__(self, plugin):
+       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.objects = self.get_objects(self.object_id)
+               self.objects.sort()
+
        def __repr__(self):
                return "<%s>" % self.__class__.__name__
 
@@ -410,68 +554,143 @@ class GraphTemplate(object):
        def log(self):
                return self.plugin.log
 
+       @property
+       def object(self):
+               """
+                       Shortcut to the main object
+               """
+               if len(self.objects) == 1:
+                       return self.objects[0]
+
        def _make_command_line(self, interval, format=DEFAULT_IMAGE_FORMAT,
-                       width=None, height=None):
-               args = []
+                       width=None, height=None, with_title=True, thumbnail=False):
+               args = [e for e in GRAPH_DEFAULT_ARGUMENTS]
 
-               args += 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 self.height),
-                       "--width", "%s" % (width or self.width),
+                       "--height", "%s" % (height or default_height),
+                       "--width", "%s" % (width or default_width),
                ]
 
                args += self.rrd_graph_args
 
+               # 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.append("--start")
+               args += ["--start", util.make_interval(interval)]
 
-               try:
-                       args.append(self.intervals[interval])
-               except KeyError:
-                       args.append(str(interval))
+               return args
+
+       def _add_defs(self):
+               use_prefix = len(self.objects) >= 2
+
+               args = []
+               for object in self.objects:
+                       if use_prefix:
+                               args += object.make_rrd_defs(object.id)
+                       else:
+                               args += object.make_rrd_defs()
 
                return args
 
-       def get_object_table(self, object_id):
-               return {
-                       "file" : self.plugin.get_object(object_id),
-               }
+       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)
 
-       def get_object_files(self, object_id):
-               files = {}
+                               # 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),
+                               ]
 
-               for id, obj in self.get_object_table(object_id).items():
-                       files[id] = obj.file
+               return ret
 
-               return files
+       def get_objects(self, *args, **kwargs):
+               object = self.plugin.get_object(*args, **kwargs)
+
+               if object:
+                       return [object,]
+
+               return []
+
+       def generate_graph(self, interval=None, **kwargs):
+               assert self.objects, "Cannot render graph without any objects"
+
+               # Make sure that all collected data is in the database
+               # to get a recent graph image
+               for object in self.objects:
+                       object.commit()
 
-       def generate_graph(self, object_id, interval=None, **kwargs):
                args = self._make_command_line(interval, **kwargs)
 
                self.log.info(_("Generating graph %s") % self)
-               self.log.debug("  args: %s" % args)
 
-               object_files = self.get_object_files(object_id)
+               rrd_graph = self.rrd_graph
 
-               for item in self.rrd_graph:
-                       try:
-                               args.append(item % object_files)
-                       except TypeError:
-                               args.append(item)
+               # Add DEFs for all objects
+               if not any((e.startswith("DEF:") for e in rrd_graph)):
+                       args += self._add_defs()
 
-               return self.write_graph(*args)
+               args += rrd_graph
+               args = self._add_vdefs(args)
 
-       def write_graph(self, *args):
-               # Convert all arguments to string
+               # Convert arguments to string
                args = [str(e) for e in args]
 
-               with tempfile.NamedTemporaryFile() as f:
-                       rrdtool.graph(f.name, *args)
+               for arg in args:
+                       self.log.debug("  %s" % arg)
 
-                       # Get back to the beginning of the file
-                       f.seek(0)
+               with Environment(self.timezone, self.locale.lang):
+                       graph = rrdtool.graphv("-", *args)
 
-                       # Return all the content
-                       return f.read()
+               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,
+               }