Author: croberts
Date: 2012-04-17 18:59:24 +0000 (Tue, 17 Apr 2012)
New Revision: 5308
Modified:
branches/play_plumage/mint/python/mint/plumage/session.py
Log:
First pass at changes to have cumin-report contain 2 threads. 1 thread starts at
"now" and moves back through time. The other thread will wake up every minute
and load all records that are new in the past minute.
Modified: branches/play_plumage/mint/python/mint/plumage/session.py
===================================================================
--- branches/play_plumage/mint/python/mint/plumage/session.py 2012-04-17 17:44:16 UTC (rev
5307)
+++ branches/play_plumage/mint/python/mint/plumage/session.py 2012-04-17 18:59:24 UTC (rev
5308)
@@ -13,8 +13,11 @@
log = logging.getLogger("mint.plumage.session")
-UTC_DIFF = datetime.utcnow() - datetime.now()
+UTC_DIFF = datetime.utcnow() - datetime.now() # needed because everything in mongo is
UTC
+class OSUtil(object):
+ pass
+
class PlumageSession(object):
def __init__(self, app, server_host, server_port):
self.app = app
@@ -42,6 +45,11 @@
self.server_host,
self.server_port,
cls))
+ self.threads.append(CurrentPlumageSessionThread(
+ self.app,
+ self.server_host,
+ self.server_port,
+ cls))
# Update our list, minus the black_list
self.app.classes.difference_update(black_list)
@@ -58,6 +66,11 @@
self.server_host,
self.server_port,
cls))
+ self.threads.append(CurrentPlumageSessionThread(
+ self.app,
+ self.server_host,
+ self.server_port,
+ cls))
def start(self):
log.info("Starting %s", self)
for t in self.threads:
@@ -86,9 +99,6 @@
# probably specified to us from a config option (with a corresponding
# query specification in the xml)
- class OSUtil(object):
- pass
-
itemSet = self.collection.find().sort("ts", -1)
for item in itemSet:
@@ -98,92 +108,29 @@
#required since MongoDB uses UTC for everything, here we get our
sanity back
setattr(record, name, value - UTC_DIFF)
else:
- setattr(record, name, item[name])
+ setattr(record, name, value)
obj = ObjectUpdate(self.app.model, record, self.cls)
self.app.update_thread.enqueue(obj)
-
-# If you'd rather fake the data, use the following instead
-# while True:
-# if self.stop_requested:
-# break
-# # We need to get some data, and we need to turn it into some kind
-# # of update object.
-# os = OSUtil()
-#
-# os.ts = time.time() * 1000000000
-#
-# machines = ["croberts.redhat.com", "tmckay.redhat.com",
"shadowman.redhat.com"]
-# os.m = machines[randrange(0,3)]
-# os.n = "%d@%s" % (randrange(1, 16), os.m)
-#
-# architectures = ["X86_64", "X86", "Itanium"]
-# os.ar = architectures[randrange(0,3)]
-#
-# oses = ["LINUX", "WINDOWS", "MAC"]
-# os.os = oses[randrange(0,3)]
-#
-# os.ki = randrange(0, 100)
-#
-# os.la = randrange(0, 100)
-#
-# states = ["Claimed", "Unclaimed", "Owner"]
-# os.st = states[randrange(0,3)]
-#
-# os.cpu = randrange(0,65)
-#
-# os.mem = randrange(0, 100000)
-#
-# obj = ObjectUpdate(self.app.model, os, self.cls)
-# self.app.update_thread.enqueue(obj)
-
-'''
-class MintConsole(Console):
- def __init__(self, model):
- self.model = model
-
- def brokerConnected(self, qmf_broker):
- message = "Broker %s:%i is connected"
- self.model.print_event(1, message, qmf_broker.host, qmf_broker.port)
-
- def brokerInfo(self, qmf_broker):
- message = "Broker info from %s:%i"
- self.model.print_event(1, message, qmf_broker.host, qmf_broker.port)
-
- def brokerDisconnected(self, qmf_broker):
- message = "Broker %s:%i is disconnected"
- self.model.print_event(1, message, qmf_broker.host, qmf_broker.port)
-
- def newAgent(self, qmf_agent):
- self.model.print_event(3, "Creating %s", qmf_agent)
-
- up = AgentUpdate(self.model, qmf_agent)
- self.model.app.update_thread.enqueue(up)
-
- def delAgent(self, qmf_agent):
- self.model.print_event(3, "Deleting %s", qmf_agent)
-
- up = AgentDelete(self.model, qmf_agent)
- self.model.app.update_thread.enqueue(up)
-
- def heartbeat(self, qmf_agent, timestamp):
- message = "Heartbeat from %s at %s"
- self.model.print_event(5, message, qmf_agent, timestamp)
-
- up = AgentUpdate(self.model, qmf_agent)
- self.model.app.update_thread.enqueue(up)
-
- def newPackage(self, name):
- self.model.print_event(2, "New package %s", name)
-
- def newClass(self, kind, classKey):
- self.model.print_event(2, "New class %s", classKey)
-
- def objectProps(self, broker, qmf_object):
- up = ObjectUpdate(self.model, qmf_object)
- self.model.app.update_thread.enqueue(up)
-
- def objectStats(self, broker, qmf_object):
- up = ObjectUpdate(self.model, qmf_object)
- self.model.app.update_thread.enqueue(up)
-'''
+
+ log.info("PlumageSessionThread--history: run completed")
+
+class CurrentPlumageSessionThread(PlumageSessionThread):
+ def run(self):
+ while True:
+ if self.stop_requested:
+ break
+ sleep(60)
+ itemSet = self.collection.find({"ts": {'$gt':
datetime.now() - timedelta(minutes=1) + UTC_DIFF}}).sort("ts", -1)
+
+ for item in itemSet:
+ record = OSUtil()
+ for name, value in item.iteritems():
+ if isinstance(value, datetime):
+ setattr(record, name, value - UTC_DIFF)
+ else:
+ setattr(record, name, value)
+
+ obj = ObjectUpdate(self.app.model, record, self.cls)
+ self.app.update_thread.enqueue(obj)
+ log.info("PlumageSessionThread--current: pass completed")