bwmon/model.py

   1 # -*- coding: utf-8 -*-
   2 #
   3 # Copyright 2010 Thomas Perl and Stefan Kögl. All rights reserved.
   4 #
   5 # Developed for a practical course (Large-scaled distributed computing) at the
   6 # University of Technology Vienna in the 2010 summer term.
   7 #
   8 # Redistribution and use in source and binary forms, with or without
   9 # modification, are permitted provided that the following conditions are met:
  10 #
  11 #    1. Redistributions of source code must retain the above copyright notice,
  12 #       this list of conditions and the following disclaimer.
  13 #
  14 #    2. Redistributions in binary form must reproduce the above copyright notice,
  15 #       this list of conditions and the following disclaimer in the documentation
  16 #       and/or other materials provided with the distribution.
  17 #
  18 # THIS SOFTWARE IS PROVIDED BY THE AUTHORS ``AS IS'' AND ANY EXPRESS OR IMPLIED
  19 # WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
  20 # MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
  21 # EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
  22 # INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  23 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
  24 # PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
  25 # LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
  26 # OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
  27 # ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  28 #
  29
  30 from __future__ import absolute_import
  31
  32 import time
  33
  34 class MonitorEntry(object):
  35     """Entity object for one monitoring "event"
  36
  37     This class encapsulates all data that's measured in
  38     one single monitoring event.
  39     """
  40     def __init__(self, cmdline, inbytes, outbytes, timestamp=None):
  41         """Creates a new MonitorEntry object
  42
  43         If the parameter timestamp is omitted, it will default to
  44         "now" (i.e. the measurement has taken place right now).
  45
  46         @param cmdline: The command line of the measured process
  47         @param inbytes: The current incoming total data, in bytes
  48         @param outbytes: The current outgoing total data, in bytes
  49         @param timestamp: Unix timestamp (UTC) of the event
  50         """
  51         if timestamp is None:
  52             timestamp = time.time()
  53
  54         self.cmdline = cmdline
  55         self.inbytes = int(inbytes)
  56         self.outbytes = int(outbytes)
  57         self.timestamp = timestamp
  58
  59     def __repr__(self):
  60         """Create a string representation of this entry
  61
  62         This is mostl used for debugging, and will simply
  63         return all the values in a user-readable string.
  64
  65         @return: The string representation of this object
  66         """
  67         return '<%s cmd="%s" in=%d out=%d time=%d>' % (self.__class__.__name__,
  68                 self.cmdline,
  69                 self.inbytes,
  70                 self.outbytes,
  71                 int(self.timestamp),)
  72
  73 class MonitorEntryCollection(object):
  74     """A (linear) collection of multiple MonitorEntry objects
  75
  76     This object is used by the monitor and the aggregator to
  77     combine several single MonitorEntry objects and provide
  78     a unified view on the current state of the objects.
  79     """
  80     TIMEOUT = 60
  81
  82     def __init__(self, update_frequency, get_app=lambda x: x):
  83         """Create a new MonitorEntryCollection object
  84
  85         The update_frequency should match the update frequency
  86         of the "calling object" (either a simple Monitor or the
  87         aggregator itself), as it's used for calculations that
  88         will be exposed to the calling object.
  89
  90         @param update_frequency: Update frequency (in seconds)
  91         @param get_app: Callback to determine the app from a command
  92         """
  93         self._data = []
  94         self._latest = {}
  95         self.update_frequency = update_frequency
  96         self.get_app = get_app
  97
  98     def expire(self):
  99         """Compact the internal state (expire old entries)
 100
 101         Remove all the measurement data that's too old and
 102         has already been processed. This will usually be
 103         called internally and remove all items older than
 104         the "TIMEOUT" value for this class.
 105         """
 106         cutoff = time.time() - self.TIMEOUT
 107         d = filter(lambda e: e.timestamp >= cutoff, self._data)
 108         self._data = d
 109         for key in self._latest.keys():
 110             current, previous = self._latest[key]
 111             if current is not None and current.timestamp < cutoff:
 112                 del self._latest[key]
 113
 114     def get_last_bytes(self, cmdline):
 115         """Return a (inbytes, outbytes, timestamp) for a command
 116
 117         In case no measurement for the given command is
 118         available, a (0, 0, 0) three-tuple will be returned.
 119
 120         Otherwise, the cumulative traffic data (inbytes, outbytes
 121         and a UTC unix timestamp) will be returned, as of the
 122         last measurement.
 123
 124         @return: A three-tuple or (0, 0, 0)
 125         """
 126         (current, previous) = self._latest.get(cmdline, (None, None))
 127
 128         if current is not None:
 129             return (current.inbytes, current.outbytes, current.timestamp)
 130
 131         return (0, 0, 0)
 132
 133     def get_bandwidth(self, cmdline):
 134         """Return a (inbytespersec, outbytespersec, timestamp) for a command
 135
 136         The semantics of this command are the same as the
 137         get_last_bytes() command, but it returns the CURRENT
 138         bandwidth usage (in bytes) instead of the cumulative
 139         bytes transferred.
 140
 141         When the current bandwidth usage cannot be determined,
 142         this function returns (0, 0, 0). This is the case when
 143         not enough measurements are available (< 2) or the
 144         command is not monitored at all by the installed monitors.
 145
 146         @return: A three-tuple or (0, 0, 0)
 147         """
 148         (current, previous) = self._latest.get(cmdline, (None, None))
 149         if current is not None and previous is not None:
 150             d_time = float(current.timestamp - previous.timestamp)
 151             d_in = float(current.inbytes - previous.inbytes)
 152             d_out = float(current.outbytes - previous.outbytes)
 153             return (d_in/d_time, d_out/d_time, current.timestamp)
 154         else:
 155             return (0, 0, 0)
 156
 157     def add(self, entry):
 158         """Add a new entry to this collection
 159
 160         This adds a new MonitorEntry to this collection and
 161         makes sure that it is picked up from the aggregation
 162         commands.
 163
 164         @param entry: A MonitorEntry object to be added
 165         """
 166         entry.cmdline = self.get_app(entry.cmdline)
 167         (current, previous) = self._latest.get(entry.cmdline, (None, None))
 168
 169         # throttle comparison; don't always take the lastest two
 170         #prev = current if current and (entry.timestamp - current.timestamp) else previous
 171         #self._latest[entry.cmdline] = (entry, prev)
 172
 173         self._latest[entry.cmdline] = (entry, current)
 174         self._data.append(entry)
 175
 176     def get_history(self, cmdline):
 177         """Return the bandwidth history for a given command
 178
 179         @param cmdline: The command line / app name to be queried
 180         """
 181         x = []
 182         last_inbytes = -1
 183         for e in self._data:
 184             if e.cmdline == cmdline:
 185                 if last_inbytes == -1:
 186                     last_inbytes = e.inbytes
 187                 x.append(e.inbytes-last_inbytes)
 188                 last_inbytes = e.inbytes
 189         return x
 190
 191     def get_datapoints(self):
 192         """Returns values for the visualization through the built-in HTTP  server
 193
 194         @return: list of x (time) and y (bandwidth usage) values;
 195         """
 196         items = [self.get_history(cmdline) for bin, bout, cmdline in self.get_usage()]
 197         if not items:
 198             return [], []
 199
 200         def check_items(l):
 201             return max(l) > 20.
 202         items = filter(check_items, items)
 203
 204         return range(len(min(items, key=len))), items
 205
 206     def get_traffic(self):
 207         """Get the current view on the traffic
 208
 209         This returns a generator object that will yield
 210         (bytes_in, bytes_out, cmdline) tuples describing
 211         the current traffic on a per-command/-app basis.
 212
 213         @return: Generator yielding (bytes_in, bytes_out, cmdline)
 214         """
 215         for cmdline in self._latest:
 216             bytes_in, bytes_out, timestamp = self.get_last_bytes(cmdline)
 217             yield (bytes_in, bytes_out, cmdline)
 218
 219     def get_usage(self):
 220         """Get the current view on the bandwidth (usage)
 221
 222         This returns a generator object that will yield
 223         (bytes_in, bytes_out, cmdline) tuples describing
 224         the current bandwidth usage on a per-command/-app
 225         basis.
 226
 227         @return: Generator yielding (bytes_in, bytes_out, cmdline)
 228         """
 229         cutoff = time.time() - self.update_frequency
 230         for cmdline in self._latest:
 231             bytes_in, bytes_out, timestamp = self.get_bandwidth(cmdline)
 232             if timestamp > cutoff:
 233                 yield (bytes_in, bytes_out, cmdline)
 234