text: import os
  import sys
  import time
  from common.stoppable_thread import StoppableThread
  
  import logging
  logger = logging.getLogger('recd')
  
  from mysql import *
  
  # Stat()
  #  Reads disk statistics from /sys/block/<dev>/stat
  #  The format of the statistics in that file are as follows:
  
      # Field  1 -- # of reads completed
      #     This is the total number of reads completed successfully.
      # Field  2 -- # of reads merged,
      #     Reads and writes which are adjacent to each other may be merged for
      #     efficiency.  Thus two 4K reads may become one 8K read before it is
      #     ultimately handed to the disk, and so it will be counted (and queued)
      #     as only one I/O.  This field lets you know how often this was done.
      # Field  3 -- # of sectors read
      #     This is the total number of sectors read successfully.
      # Field  4 -- # of milliseconds spent reading
      #     This is the total number of milliseconds spent by all reads (as
      #     measured from __make_request() to end_that_request_last()).
      # Field  5 -- # of writes completed
      #     This is the total number of writes completed successfully.
      # Field 6 -- # of writes merged
      # Field  7 -- # of sectors written
      #     This is the total number of sectors written successfully.
      # Field  8 -- # of milliseconds spent writing
      #     This is the total number of milliseconds spent by all writes (as
      #     measured from __make_request() to end_that_request_last()).
      # Field  9 -- # of I/Os currently in progress
      #     The only field that should go to zero. Incremented as requests are
      #     given to appropriate struct request_queue and decremented as they finish.
      # Field 10 -- # of milliseconds spent doing I/Os
      #     This field increases so long as field 9 is nonzero.
      # Field 11 -- weighted # of milliseconds spent doing I/Os
      #     This field is incremented at each I/O start, I/O completion, I/O
      #     merge, or read of these stats by the number of I/Os in progress
      #     (field 9) times the number of milliseconds spent doing I/O since the
      #     last update of this field.  This can provide an easy measure of both
      #     I/O completion time and the backlog that may be accumulating.
  class Stat(object):
      __db_table__ = 'disk_stats'
      __db_columns__ = ['device',
                        'read_complete',
                        'read_merged',
                        'read_sectors',
                        'read_ms',
                        'write_complete',
                        'write_merged',
                        'write_sectors',
                        'write_ms',
                        'ios_in_prog',
                        'io_ms',
                        'weighted_io_ms',
                        'read_bytes',
                        'write_bytes',
                        'utilization']
  
      READ_COMPLETE   = 0
      READ_MERGED     = 1
      READ_SECTORS    = 2
      READ_MS         = 3
      WRITE_COMPLETE  = 4
      WRITE_MERGED    = 5
      WRITE_SECTORS   = 6
      WRITE_MS        = 7
      IOS_IN_PROG     = 8
      IO_MS           = 9
      WEIGHTED_IO_MS  = 10
  
      def __init__(self,disk):
          self.disk = disk
          self.disk_ok = os.path.exists('/sys/block/'+disk+'/stat')
          self.last = None
          self.last_time = None
          if not self.disk_ok:
              logger.warning('Disk %s does not exist. Ignoring.' % disk)
              return
  
          # Find the sector size to determine the number of bytes read/written
          self.sector_size = int(file('/sys/block/'+self.disk+'/queue/hw_sector_size','r').read())
          logger.info(disk + ': Sector Size is %d bytes.' % self.sector_size)
  
      def called(self):
          logger.info(self.disk + ' called')
  
      def update(self):
          stat = file('/sys/block/'+self.disk+'/stat','r').read()
          cols = stat.split()
          read_time = time.time()
  
          # Startup
          if self.last is None:
              self.last_time = read_time
              self.last = cols
              return
  
          # Use time difference to account for processing and sleep resolution
          diff_time = read_time - self.last_time
  
          # Build a list of differences
          self.diff = [ int((int(col)-int(last))/diff_time) for last,col in zip(self.last,cols) ]
  
          # IOs in progress is a queue length so it can go to zero. Do not subtract it
          self.diff[Stat.IOS_IN_PROG] = int(cols[Stat.IOS_IN_PROG])
  
          self.last_time = read_time
          self.last = cols
  
      def insert(self,options):
          vals = []
  
          # Disk Name
          vals.append(self.disk)
          # The 11 values which are diffed with the previous second
          vals.extend(self.diff)
  
          # Caluclated Columns
          # read_bytes
          vals.append(self.diff[Stat.READ_SECTORS]*self.sector_size)
          # write_bytes
          vals.append(self.diff[Stat.WRITE_SECTORS]*self.sector_size)
          # utilization - max 100%
          vals.append('%3.2f' % min((self.diff[Stat.IO_MS]/10.0),100.00))
  
          # Insert into the db
          db_insert_into(options,Stat.__db_table__,Stat.__db_columns__,vals)
  
  
  class DiskStatsCleaner(StoppableThread):
      def __init__(self,options):
          super(DiskStatsCleaner, self).__init__()
          self.options = options
  
      def clean(self):
          # Delete rows where the time is < the history amount
          sql_where = "time < DATE_SUB(CURRENT_TIMESTAMP, INTERVAL %d HOUR)" % (self.options.disk_stats_history)
          rows_deleted = db_delete_from(self.options,Stat.__db_table__,where=sql_where)
          #logger.info("Cleaned: %d" % rows_deleted)
  
      def __run__(self):
          logger.info("DiskStatsCleaner Started")
          loop = 0
  
          while not self.stopped():
              time.sleep(1)
              loop+=1
              # Check every minute
              if loop % 60 == 0:
                  self.clean()
  
          logger.info("DiskStatsCleaner Stopped")
  
  class DiskStats(StoppableThread):
      def __init__(self, options):
          super(DiskStats, self).__init__()
          self.options = options
          self.cleaner = DiskStatsCleaner(options)
  
      def __run__(self):
          logger.info("DiskStats Started")
          self.cleaner.start()
  
          disks = [Stat(device) for device in self.options.disk_stats_devices]
  
          # Filter out disks we can't track
          disks = filter(lambda d: d.disk_ok, disks)
  
          # Exit if no devices to track
          if len(disks) is 0:
              logger.warning("No devices to track. Exiting")
              self.cleaner.stop()
              self.cleaner.join()
              return
  
          # Call update one on all first so they can get initial values
          for disk in disks: disk.update()
  
          # Break if the user stops
          to_sleep = 1
          while not self.stopped(timeout=to_sleep):
              last = time.time()
  
              # Split up the update from the insert
              # This tries to get a snapshot of all the disks at the same time
              for disk in disks: disk.update()
              for disk in disks: disk.insert(self.options)
  
              # Sleep the shortest amount of time possible so we get updates on the second boundary
              to_sleep = 1-(time.time()-last)
              if to_sleep < 0: to_sleep = 0
  
          self.cleaner.stop()
          self.cleaner.join()
          self.cleaner.raise_exception(log=logger)
  
          logger.info("DiskStats has stopped")
  
  
  # Empty object that you can add fields to.
  class Object(object):
      pass
  
  
  if __name__ == "__main__":
      # print to stdout
      logger.setLevel(logging.DEBUG)
      formatter = logging.Formatter("%(asctime)s  %(levelname)s %(name)s: %(funcName)s::%(filename)s:%(lineno)s  %(message)s")
      streamHandler = logging.StreamHandler()
      streamHandler.setFormatter(formatter)
      logger.addHandler(streamHandler)
  
      logger.info('DiskStats Unit Testing')
  
      # Fake options for testing
      options = Object()
      options.dbhostname = 'localhost'
      options.dbusername = 'root'
      options.dbpassword = 'gizzy'
      options.dbname = 'recd'
      options.unit_test = False
      options.disk_stats_devices = ['sda','sdb','sdc','sdd','sde']
      options.disk_stats_history = 48 # hours
  
      ds = DiskStats(options)
      ds.start()
  
      # Stop all threads for ctrl-c
      while ds.isAlive():
          try:
              time.sleep(1)
          except (KeyboardInterrupt, SystemExit):
              break
  
      logger.info('stop')
      ds.stop()
      logger.info('join')
      ds.join()
      logger.info('end')
      ds.raise_exception(log=logger)
  