diff --git a/wally/sensors_utils.py b/wally/sensors_utils.py
index 51a4dcb..e5b4347 100644
--- a/wally/sensors_utils.py
+++ b/wally/sensors_utils.py
@@ -1,4 +1,5 @@
 import time
+import array
 import Queue
 import logging
 import threading
@@ -11,14 +12,91 @@
                                stop_and_remove_sensors)
 
 
-logger = logging.getLogger("wally")
+logger = logging.getLogger("wally.sensors")
 DEFAULT_RECEIVER_URL = "udp://{ip}:5699"
 
 
-def save_sensors_data(data_q, mon_q, fd):
+class SensorDatastore(object):
+    def __init__(self, stime=None):
+        self.lock = threading.Lock()
+        self.stime = stime
+
+        self.min_size = 60 * 60
+        self.max_size = 60 * 61
+
+        self.data = {
+            'testnodes:io': array.array("B"),
+            'testnodes:cpu': array.array("B"),
+        }
+
+    def get_values(self, name, start, end):
+        assert end >= start
+        if end == start:
+            return []
+
+        with self.lock:
+            curr_arr = self.data[name]
+            if self.stime is None:
+                return []
+
+            sidx = start - self.stime
+            eidx = end - self.stime
+
+            if sidx < 0 and eidx < 0:
+                return [0] * (end - start)
+            elif sidx < 0:
+                return [0] * (-sidx) + curr_arr[:eidx]
+            return curr_arr[sidx:eidx]
+
+    def set_values(self, start_time, vals):
+        with self.lock:
+            return self.add_values_l(start_time, vals)
+
+    def set_values_l(self, start_time, vals):
+        max_cut = 0
+        for name, values in vals.items():
+            curr_arr = self.data.setdefault(name, array.array("H"))
+
+            if self.stime is None:
+                self.stime = start_time
+
+            curr_end_time = len(curr_arr) + self.stime
+
+            if curr_end_time < start_time:
+                curr_arr.extend([0] * (start_time - curr_end_time))
+                curr_arr.extend(values)
+            elif curr_end_time > start_time:
+                logger.warning("Duplicated sensors data")
+                sindex = len(curr_arr) + (start_time - curr_end_time)
+
+                if sindex < 0:
+                    values = values[-sindex:]
+                    sindex = 0
+                    logger.warning("Some data with timestamp before"
+                                   " beginning of measurememts. Skip them")
+
+                curr_arr[sindex:sindex + len(values)] = values
+            else:
+                curr_arr.extend(values)
+
+            if len(curr_arr) > self.max_size:
+                max_cut = max(len(curr_arr) - self.min_size, max_cut)
+
+        if max_cut > 0:
+            self.start_time += max_cut
+            for val in vals.values():
+                del val[:max_cut]
+
+
+def save_sensors_data(data_q, mon_q, fd, data_store, source2roles_map):
     fd.write("\n")
 
+    BUFFER = 3
     observed_nodes = set()
+    testnodes_data = {
+        'io': {},
+        'cpu': {},
+    }
 
     try:
         while True:
@@ -32,8 +110,27 @@
                 mon_q.put(addr + (data['source_id'],))
                 observed_nodes.add(addr)
 
-            fd.write("{0!s} : {1!r}\n".format(time.time(),
-                     repr((addr, data))))
+            fd.write(repr((addr, data)) + "\n")
+
+            source_id = data.pop('source_id')
+            rep_time = data.pop('time')
+            if 'testnode' in source2roles_map.get(source_id, []):
+                vl = testnodes_data['io'].get(rep_time, 0)
+                sum_io_q = vl
+                testnodes_data['io'][rep_time] = sum_io_q
+
+            etime = time.time() - BUFFER
+            for name, vals in testnodes_data.items():
+                new_vals = {}
+                for rtime, value in vals.items():
+                    if rtime < etime:
+                        data_store.set_values("testnodes:io", rtime, [value])
+                    else:
+                        new_vals[rtime] = value
+
+                vals.clear()
+                vals.update(new_vals)
+
     except Exception:
         logger.exception("Error in sensors thread")
     logger.info("Sensors thread exits")
@@ -42,6 +139,7 @@
 def get_sensors_config_for_nodes(cfg, nodes):
     monitored_nodes = []
     sensors_configs = []
+    source2roles_map = {}
 
     receiver_url = cfg.get("receiver_url", DEFAULT_RECEIVER_URL)
     assert '{ip}' in receiver_url
@@ -64,6 +162,7 @@
                         ext_ip = utils.get_ip_for_target(ip)
                     monitor_url = receiver_url.format(ip=ext_ip)
 
+                source2roles_map[node.get_conn_id()] = node.roles
                 monitored_nodes.append(node)
                 sens_cfg = SensorConfig(node.connection,
                                         node.get_conn_id(),
@@ -72,18 +171,20 @@
                                         monitor_url=monitor_url)
                 sensors_configs.append(sens_cfg)
 
-    return monitored_nodes, sensors_configs
+    return monitored_nodes, sensors_configs, source2roles_map
 
 
-def start_sensor_process_thread(ctx, cfg, sensors_configs):
+def start_sensor_process_thread(ctx, cfg, sensors_configs, source2roles_map):
     receiver_url = cfg.get('receiver_url', DEFAULT_RECEIVER_URL)
     sensors_data_q, stop_sensors_loop = \
         start_listener_thread(receiver_url.format(ip='0.0.0.0'))
 
     mon_q = Queue.Queue()
     fd = open(cfg_dict['sensor_storage'], "w")
+
+    params = sensors_data_q, mon_q, fd, ctx.sensors_data, source2roles_map
     sensor_listen_th = threading.Thread(None, save_sensors_data, None,
-                                        (sensors_data_q, mon_q, fd))
+                                        params)
     sensor_listen_th.daemon = True
     sensor_listen_th.start()
 
@@ -105,8 +206,8 @@
     if nodes is None:
         nodes = ctx.nodes
 
-    monitored_nodes, sensors_configs = get_sensors_config_for_nodes(cfg,
-                                                                    nodes)
+    monitored_nodes, sensors_configs, source2roles_map = \
+        get_sensors_config_for_nodes(cfg, nodes)
 
     if len(monitored_nodes) == 0:
         logger.info("Nothing to monitor, no sensors would be installed")
@@ -115,12 +216,13 @@
     if ctx.sensors_mon_q is None:
         logger.info("Start sensors data receiving thread")
         ctx.sensors_mon_q = start_sensor_process_thread(ctx, cfg,
-                                                        sensors_configs)
+                                                        sensors_configs,
+                                                        source2roles_map)
 
     if undeploy:
         def remove_sensors_stage(cfg, ctx):
-            _, sensors_configs = get_sensors_config_for_nodes(cfg['sensors'],
-                                                              nodes)
+            _, sensors_configs, _ = \
+                get_sensors_config_for_nodes(cfg['sensors'], nodes)
             stop_and_remove_sensors(sensors_configs)
 
         ctx.clear_calls_stack.append(remove_sensors_stage)