From fb7de0c03671c544bb51bae4701f462ef994e684 Mon Sep 17 00:00:00 2001
From: Christian Pointner <equinox@spreadspace.org>
Date: Mon, 25 Aug 2014 22:16:57 +0200
Subject: rrd importer works now

---
 src/daq/flumotion-rrd/flumotion-rrd.py | 54 +++++++++++++++++++++++++++-------
 1 file changed, 44 insertions(+), 10 deletions(-)

(limited to 'src/daq/flumotion-rrd/flumotion-rrd.py')

diff --git a/src/daq/flumotion-rrd/flumotion-rrd.py b/src/daq/flumotion-rrd/flumotion-rrd.py
index 71d3257..405b335 100755
--- a/src/daq/flumotion-rrd/flumotion-rrd.py
+++ b/src/daq/flumotion-rrd/flumotion-rrd.py
@@ -98,20 +98,33 @@ class FlumotionRRD():
 
     def _socketReady(self):
         print 'SFive: connection to sfive hub established'
-        self._importRRD()
+        self._checkRRD()
         self._sendInit()
-        # TODO: start data import
-        self._sendDataset(datetime.datetime.utcnow(), 5, 17, 1214, 14214)
-        print 'SFive: sent %d datasets' % (1)
+        cnt = self._sendRRD()
+        print 'SFive: sent %d datasets' % (cnt)
 
         reactor.stop()
 
-    def _importRRD(self):
-        print "bytes:\n"
-        print rrdtool.info(self._bytes_rrdfile)
+    def _checkRRD(self):
+        info_bytes = rrdtool.info(self._bytes_rrdfile)
+        step_bytes = info_bytes['step']
+        lastupdate_bytes = info_bytes['last_update']
+        info_clients = rrdtool.info(self._clients_rrdfile)
+        step_clients = info_clients['step']
+        lastupdate_clients = info_clients['last_update']
 
-        print "\n\nclients:\n"
-        print rrdtool.info(self._clients_rrdfile)
+        self._duration = step_bytes
+        if step_bytes != step_clients:
+            print 'SFive: ERROR step size of the RRD Files don\'t match'
+            reactor.stop()
+
+        self._end = lastupdate_bytes if lastupdate_bytes < lastupdate_clients else lastupdate_clients
+        if lastupdate_bytes != lastupdate_clients:
+            print 'SFive: WARNING the last update timestamp of the RRD Files don\'t match - will use smaller timestamp'
+            reactor.stop()
+
+        print 'SFive: will use 7 days of data ending with %s using a step size of %d seconds' % (
+                   datetime.datetime.utcfromtimestamp(self._end).isoformat('Z'), self._duration)
 
     def _sendInit(self):
         initdata = { "hostname": self._hostname,
@@ -119,7 +132,28 @@ class FlumotionRRD():
                      "tags": self._tags }
         self._proto.sendDatagram('%s\n' % (json.dumps(initdata)));
 
-    def _sendDataset(self, timestamp, duration, client_count, bytes_received, bytes_sent):
+    def _sendRRD(self):
+        data_bytes = rrdtool.fetch(self._bytes_rrdfile, 'AVERAGE', '-s', 'end-7d', '-e', '%d' % self._end)
+        data_clients = rrdtool.fetch(self._clients_rrdfile, 'AVERAGE', '-s', 'end-7d', '-e', '%d' % self._end)
+
+        ts = data_bytes[0][0]
+        max = len(data_bytes[2])
+        if max < 1:
+            return 0
+
+        cnt = 0
+        for idx in range(0, max - 1):
+            try:
+                self._sendDataset(datetime.datetime.utcfromtimestamp(ts), self._duration,
+                                  data_clients[2][idx][0], data_bytes[2][idx][0])
+                ts += self._duration
+                cnt += 1
+            except ValueError as err:
+                print 'SFive: timestamp "%d" seems wrong: %s' % (ts, err)
+
+        return max
+
+    def _sendDataset(self, timestamp, duration, client_count, bytes_sent):
         data = { "start-time": timestamp.isoformat('Z'),
                  "duration-ms": duration * 1000,
                  "data": {
-- 
cgit v1.2.3