Author: croberts
Date: 2012-12-07 18:00:35 +0000 (Fri, 07 Dec 2012)
New Revision: 5567
Modified:
trunk/mint/python/mint/plumage/session.py
Log:
BZ 883784: Addressing this by eliminating the map_reduce logic that tallied cpu/memory in
favor of just looping through the cursor for each type of slot (used, unused, etc) at each
time period. This approach seems less flaky and should still yield proper results.
Modified: trunk/mint/python/mint/plumage/session.py
===================================================================
--- trunk/mint/python/mint/plumage/session.py 2012-12-06 21:18:00 UTC (rev 5566)
+++ trunk/mint/python/mint/plumage/session.py 2012-12-07 18:00:35 UTC (rev 5567)
@@ -7,7 +7,6 @@
from update import ObjectUpdate
from random import randrange
from datetime import timedelta, datetime
-from bson.code import Code
try:
from pymongo import Connection
@@ -24,28 +23,19 @@
UTC_DIFF = datetime.utcnow() - datetime.now() # needed because everything in mongo is
UTC
-timestamp_mapper = Code("""
- function () {
- emit({ts:this.ts, st:this.st}, {mem: this.mem, cpu: this.cpu});
- }
- """)
-timestamp_reducer = Code("""
- function (key, values) {
- var memsum = 0;
- var cpusum = 0;
- values.forEach(function(doc) {
- memsum += doc.mem;
- cpusum += doc.cpu;
- });
- return {mem: memsum, cpu: cpusum}; //format of this needs to be
same as second param of emit
- }
- """)
-
-
class RecordObject(object):
- pass
+ def __str__(self):
+ sb = []
+ for key in self.__dict__:
+ sb.append("{key}='{value}'".format(key=key,
value=self.__dict__[key]))
+
+ return ', '.join(sb)
+
+ def __repr__(self):
+ return self.__str__()
+
class PlumageSession(object):
def __init__(self, app, server_host, server_port):
self.app = app
@@ -302,43 +292,47 @@
class PlumageOSUtilSessionThread(PlumageSessionThread):
def fillOSUtilStats(self, time, temptable):
record = RecordObject()
+
+ totalrecs = self.collection.find({'ts':time})
+ usedrecs =
self.collection.find({'ts':time,'st':{'$nin':['Unclaimed','Owner']}})
+ unusedrecs =
self.collection.find({'ts':time,'st':'Unclaimed'})
+ ownerrecs =
self.collection.find({'ts':time,'st':'Owner'})
+ matchedrecs =
self.collection.find({'ts':time,'st':'Matched'})
+
record.host= "%s:%s" % (self.connection.host, self.connection.port)
- record.total =
len(self.collection.find({'ts':time}).distinct('n'))
- record.used =
len(self.collection.find({'ts':time,'st':{'$nin':['Unclaimed','Owner']}}).distinct('n'))
- record.unused =
len(self.collection.find({'ts':time,'st':'Unclaimed'}).distinct('n'))
- record.owner =
len(self.collection.find({'ts':time,'st':'Owner'}).distinct('n'))
- reduced = self.collection.map_reduce(timestamp_mapper, timestamp_reducer,
temptable, query={'ts':time,})
- reduced_collection = Collection(self.database,temptable)
+ record.total = len(totalrecs.distinct('n'))
+ record.used = len(usedrecs.distinct('n'))
+ record.unused = len(unusedrecs.distinct('n'))
+ record.owner = len(ownerrecs.distinct('n'))
- result_claimed = reduced_collection.find({'_id.st':'Claimed'})
- result_unclaimed =
reduced_collection.find({'_id.st':'Unclaimed'})
- result_owner = reduced_collection.find({'_id.st':'Owner'})
- result_matched = reduced_collection.find({'_id.st':'Matched'})
-
- record.freemem = self.getStat(result_unclaimed, 'mem')
- record.freecpu = self.getStat(result_unclaimed, 'cpu')
- record.usedmem = self.getStat(result_claimed, 'mem')
- record.usedcpu = self.getStat(result_claimed, 'cpu')
- record.availmem = self.getStat(result_claimed, 'mem') +
self.getStat(result_unclaimed, 'mem')
- record.availcpu = self.getStat(result_claimed, 'cpu') +
self.getStat(result_unclaimed, 'cpu')
- record.totalmem = self.getStat(result_claimed, 'mem') +
self.getStat(result_unclaimed, 'mem') + self.getStat(result_owner, 'mem')
+ self.getStat(result_matched, 'mem')
- record.totalcpu = self.getStat(result_claimed, 'cpu') +
self.getStat(result_unclaimed, 'cpu') + self.getStat(result_owner, 'cpu')
+ self.getStat(result_matched, 'cpu')
+ record.freemem = self.getStat(unusedrecs, 'mem')
+ record.freecpu = self.getStat(unusedrecs, 'cpu')
+ record.usedmem = self.getStat(usedrecs, 'mem')
+ record.usedcpu = self.getStat(usedrecs, 'cpu')
+ record.availmem = self.getStat(usedrecs, 'mem') +
self.getStat(unusedrecs, 'mem')
+ record.availcpu = self.getStat(usedrecs, 'cpu') +
self.getStat(unusedrecs, 'cpu')
+ record.totalmem = self.getStat(usedrecs, 'mem') +
self.getStat(unusedrecs, 'mem') + self.getStat(ownerrecs, 'mem') +
self.getStat(matchedrecs, 'mem')
+ record.totalcpu = self.getStat(usedrecs, 'cpu') +
self.getStat(unusedrecs, 'cpu') + self.getStat(ownerrecs, 'cpu') +
self.getStat(matchedrecs, 'cpu')
if(record.total != record.owner):
record.efficiency = (record.used/(record.total-record.owner))*100
else:
record.efficiency = 0
record.ts = time - UTC_DIFF
+
+ log.debug("For time [%s] Adding OSUtil record of [%s]" %
(time,record))
return record
- def getStat(self, result_object, item):
- itemcount = 0
- try:
- itemcount = result_object[0]['value'][item]
- except:
- return 0
- return itemcount
-
+ def getStat(self, recordset, stat):
+ recordset.rewind() # be sure that we're working with a fresh cursor
+ total = 0
+ for rec in recordset:
+ try:
+ total += rec[stat]
+ except:
+ pass
+ return total
+
def run(self):
while True:
self._check_connection()