Take running time of plugins into account when calculating interval.

[collecty.git] / collecty / plugins / base.py

diff --git a/collecty/plugins/base.py b/collecty/plugins/base.py
index ddb5e164ca5df1dc6af9c5e465b32cbf41adb096..2997d11d74a05cc2931950e53b95672ee3bfd997 100644 (file)
--- a/collecty/plugins/base.py
+++ b/collecty/plugins/base.py
@@ -36,18 +36,22 @@ class Timer(object):
                self.timeout = timeout
                self.heartbeat = heartbeat
 
+               self.delay = 0
+
                self.reset()
 
-       def reset(self):
+       def reset(self, delay=0):
                # Save start time.
                self.start = time.time()
 
+               self.delay = delay
+
                # Has this timer been killed?
                self.killed = False
 
        @property
        def elapsed(self):
-               return time.time() - self.start
+               return time.time() - self.start - self.delay
 
        def cancel(self):
                self.killed = True
@@ -135,6 +139,10 @@ class DataSource(threading.Thread):
        def stepsize(self):
                return self.interval
 
+       @property
+       def heartbeat(self):
+               return self.stepsize * 2
+
        @property
        def file(self):
                """
@@ -155,13 +163,13 @@ class DataSource(threading.Thread):
                        os.makedirs(dirname)
 
                # Create argument list.
-               args = [
-                       "--step", "%s" % self.default_interval,
-               ] + self.get_rrd_schema()
+               args = self.get_rrd_schema()
 
                rrdtool.create(self.file, *args)
 
                self.log.debug(_("Created RRD file %s.") % self.file)
+               for arg in args:
+                       self.log.debug("  %s" % arg)
 
        def get_rrd_schema(self):
                schema = [
@@ -176,7 +184,7 @@ class DataSource(threading.Thread):
                                                prefix,
                                                name,
                                                type,
-                                               "%s" % self.stepsize,
+                                               "%s" % self.heartbeat,
                                                lower_limit,
                                                upper_limit
                                        ))
@@ -231,6 +239,13 @@ class DataSource(threading.Thread):
                        return
 
                self.log.debug(_("Submitting data to database. %d entries.") % len(self.data))
+               for data in self.data:
+                       self.log.debug("  %s" % data)
+
+               # Create the RRD files (if they don't exist yet or
+               # have vanished for some reason).
+               self.create()
+
                rrdtool.update(self.file, *self.data)
                self.data = []
 
@@ -238,13 +253,22 @@ class DataSource(threading.Thread):
                """
                        This method catches errors from the read() method and logs them.
                """
+               start_time = time.time()
+
                try:
-                       return self.read(*args, **kwargs)
+                       data = self.read(*args, **kwargs)
+                       if data is None:
+                               self.log.warning(_("Received empty data."))
+                       else:
+                               self.data.append("%d:%s" % (start_time, data))
 
                # Catch any exceptions, so collecty does not crash.
                except Exception, e:
                        self.log.critical(_("Unhandled exception in read()!"), exc_info=True)
 
+               # Return the elapsed time since _read() has been called.
+               return (time.time() - start_time)
+
        def _submit(self, *args, **kwargs):
                """
                        This method catches errors from the submit() method and logs them.
@@ -266,7 +290,9 @@ class DataSource(threading.Thread):
                        # Wait until the timer has successfully elapsed.
                        if self.timer.wait():
                                self.log.debug(_("Collecting..."))
-                               self._read()
+                               delay = self._read()
+
+                               self.timer.reset(delay)
 
                self._submit()
                self.log.debug(_("Stopped."))