Author: tmckay
Date: 2012-04-20 19:55:11 +0000 (Fri, 20 Apr 2012)
New Revision: 5333
Modified:
trunk/mint/python/mint/plumage/session.py
Log:
Add handlers for failed import, mongo server not up, general errors for graceful failure.
Modified: trunk/mint/python/mint/plumage/session.py
===================================================================
--- trunk/mint/python/mint/plumage/session.py 2012-04-20 18:49:12 UTC (rev 5332)
+++ trunk/mint/python/mint/plumage/session.py 2012-04-20 19:55:11 UTC (rev 5333)
@@ -1,5 +1,4 @@
from __future__ import division
-
import time
from mint.util import MintDaemonThread
@@ -8,12 +7,15 @@
from random import randrange
from datetime import timedelta, datetime
-from pymongo import Connection
-from pymongo.collection import Collection
-from pymongo.database import Database
+try:
+ from pymongo import Connection
+ from pymongo.errors import AutoReconnect
+ from pymongo.collection import Collection
+ from pymongo.database import Database
+ imports_ok = True
+except:
+ imports_ok = False
-from dateutil.parser import *
-
import logging
log = logging.getLogger("mint.plumage.session")
@@ -37,6 +39,9 @@
log.info("Initializing %s", self)
def init_classes(self):
+ if not imports_ok:
+ return
+
# Apply the package filter to the class list
if len(self.app.classes):
black_list = set()
@@ -88,6 +93,9 @@
cls))
def start(self):
log.info("Starting %s", self)
+ if not imports_ok:
+ log.info("!!! Imports failed for pymongo, there will be no data
!!!")
+
for t in self.threads:
t.start()
@@ -104,58 +112,68 @@
super(PlumageSessionThread, self).__init__(app)
self.cls = cls
- self.connection = Connection(server_host, server_port) #Needs to be from config
- self.database = Database(self.connection, self.cls.database)
- self.collection = Collection(self.database, self.cls.collection)
+ self.server_host = server_host
+ self.server_port = server_port
+
+ try:
+ self.connection = Connection(server_host, server_port)
+ except AutoReconnect:
+ self.connection = None
+ log.info("%s could not connect to mongo server at %s:%s, "\
+ "autoreconnect is on" % (self.__class__.__name__,
+ server_host,
+ server_port))
+ self.database = None
+ self.collection = None
+
+ def _init(self):
+ if self.database is None:
+ self.database = Database(self.connection, self.cls.database)
+
+ if self.collection is None:
+ self.collection = Collection(self.database, self.cls.collection)
+
+ def _check_connection(self):
+ # Loop waiting for the server if it's not there. Autoreconnect
+ # will hook us up...
+ while not self.stop_requested:
+ try:
+ if self.connection is None:
+ self.connection = Connection(self.server_host,
+ self.server_port)
+ else:
+ self.connection.server_info()
+ break
+ except AutoReconnect, e:
+ log.info("%s waiting for server, %s, sleeping..."\
+ % (self.__class__.__name__, str(e)))
+ for i in range(10):
+ sleep(1)
+ if self.stop_requested:
+ break
def run(self):
+ self._check_connection()
+ if self.stop_requested:
+ return
- # We create objects here. Tag them with the right class,
- # probably specified to us from a config option (with a corresponding
- # query specification in the xml)
- (oldest, newest) =
self.app.update_thread.get_first_and_last_sample_timestamp(self.cls)
- if oldest is None:
- # if we have no oldest record (first run), start at "5 min ago" and
start loading everything
- oldest = datetime.now() - timedelta(seconds=300)
- oldest = oldest + UTC_DIFF
-
- log.info("PlumageSessionThread--history: Loading records older than
%s" % oldest)
- times = sorted(self.collection.find({"ts": {'$lt':
oldest}}).distinct('ts'), reverse=True)
- sample_times = map(lambda i: times[i],filter(lambda i: i%5 ==
0,range(len(times))))
+ try:
+ # Make sure we have a database
+ self._init()
- for time in sample_times:
- record = OSUtil()
- 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'))
- record.efficiency = (record.used/(record.total-record.owner))*100
- record.ts = time - UTC_DIFF
-
- obj = ObjectUpdate(self.app.model, record, self.cls)
- self.app.update_thread.enqueue(obj)
-
- log.info("PlumageSessionThread--history: run completed")
-
-
-class CurrentPlumageSessionThread(PlumageSessionThread):
- def run(self):
- while True:
- if self.stop_requested:
- break
-
+ # We create objects here. Tag them with the right class,
+ # probably specified to us from a config option (with a corresponding
+ # query specification in the xml)
(oldest, newest) =
self.app.update_thread.get_first_and_last_sample_timestamp(self.cls)
- if newest is not None:
- most_recent = max((datetime.now() - timedelta(seconds=300) + UTC_DIFF),
(newest + UTC_DIFF))
- else:
- most_recent = datetime.now() - timedelta(seconds=300) + UTC_DIFF
+ if oldest is None:
+ # if we have no oldest record (first run), start at "5 min ago"
and start loading everything
+ oldest = datetime.now() - timedelta(seconds=300)
+ oldest = oldest + UTC_DIFF
- log.info("PlumageSessionThread--current: Loading records newer than
%s" % most_recent)
+ log.info("PlumageSessionThread--history: Loading records older than
%s" % oldest)
+ times = sorted(self.collection.find({"ts": {'$lt':
oldest}}).distinct('ts'), reverse=True)
+ sample_times = map(lambda i: times[i],filter(lambda i: i%5 ==
0,range(len(times))))
- times = sorted(self.collection.find({"ts": {'$gt':
most_recent}}).distinct('ts'), reverse=True)
- sample_times = map(lambda i: times[i],filter(lambda i: i%5 ==
0,range(len(times))))
-
for time in sample_times:
record = OSUtil()
record.host= "%s:%s" % (self.connection.host,
self.connection.port)
@@ -165,10 +183,53 @@
record.owner =
len(self.collection.find({'ts':time,'st':'Owner'}).distinct('n'))
record.efficiency = (record.used/(record.total-record.owner))*100
record.ts = time - UTC_DIFF
-
+
obj = ObjectUpdate(self.app.model, record, self.cls)
self.app.update_thread.enqueue(obj)
+
+ log.info("PlumageSessionThread--history: run completed")
+
+ except Exception, e:
+ log.info("%s got exception %s, exiting" % (self.__class__.__name__,
str(e)))
+
+class CurrentPlumageSessionThread(PlumageSessionThread):
+ def run(self):
+ while True:
+ self._check_connection()
+ if self.stop_requested:
+ break
+
+ # Make sure we have a db
+ try:
+ self._init()
+
+ (oldest, newest) =
self.app.update_thread.get_first_and_last_sample_timestamp(self.cls)
+ if newest is not None:
+ most_recent = max((datetime.now() - timedelta(seconds=300) +
UTC_DIFF), (newest + UTC_DIFF))
+ else:
+ most_recent = datetime.now() - timedelta(seconds=300) + UTC_DIFF
+
+ log.info("PlumageSessionThread--current: Loading records newer than
%s" % most_recent)
+
+ times = sorted(self.collection.find({"ts": {'$gt':
most_recent}}).distinct('ts'), reverse=True)
+ sample_times = map(lambda i: times[i],filter(lambda i: i%5 ==
0,range(len(times))))
+
+ for time in sample_times:
+ record = OSUtil()
+ 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'))
+ record.efficiency = (record.used/(record.total-record.owner))*100
+ record.ts = time - UTC_DIFF
+
+ obj = ObjectUpdate(self.app.model, record, self.cls)
+ self.app.update_thread.enqueue(obj)
+
log.info("PlumageSessionThread--current: pass completed")
+ except Exception, e:
+ log.info("%s got exception %s, sleeping" %
(self.__class__.__name__, str(e)))
sleep(300)
@@ -178,27 +239,37 @@
'''
class CatchUpPlumageSessionThread(PlumageSessionThread):
def run(self):
- (oldest, newest) =
self.app.update_thread.get_first_and_last_sample_timestamp(self.cls)
- if newest is not None:
- log.info("CatchUpPlumageSessionThread: Starting for records newer than
%s" % newest)
-
- times = sorted(self.collection.find({"ts": {'$gt': newest +
UTC_DIFF, '$lt': datetime.now() - timedelta(seconds=300) +
UTC_DIFF}}).distinct('ts'), reverse=True)
- sample_times = map(lambda i: times[i],filter(lambda i: i%5 ==
0,range(len(times))))
-
- for time in sample_times:
- record = OSUtil()
- 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'))
- record.efficiency = (record.used/(record.total-record.owner))*100
- record.ts = time - UTC_DIFF
-
- obj = ObjectUpdate(self.app.model, record, self.cls)
- self.app.update_thread.enqueue(obj)
- log.info("PlumageSessionThread--catch-up: pass completed")
- sleep(300)
- log.info("CatchUpPlumageSessionThread--catch-up: catch-up run completed
for records newer than %s and older than %s" % (newest, datetime.now() -
timedelta(seconds=300) + UTC_DIFF))
- else:
- log.info("CatchUpPlumageSessionThread: Skipping catch-up, no records
present (probably first-run)")
\ No newline at end of file
+ self._check_connection()
+ if self.stop_requested:
+ return
+
+ try:
+ # Make sure we have a db
+ self._init()
+
+ (oldest, newest) =
self.app.update_thread.get_first_and_last_sample_timestamp(self.cls)
+ if newest is not None:
+ log.info("CatchUpPlumageSessionThread: Starting for records newer
than %s" % newest)
+
+ times = sorted(self.collection.find({"ts": {'$gt':
newest + UTC_DIFF, '$lt': datetime.now() - timedelta(seconds=300) +
UTC_DIFF}}).distinct('ts'), reverse=True)
+ sample_times = map(lambda i: times[i],filter(lambda i: i%5 ==
0,range(len(times))))
+
+ for time in sample_times:
+ record = OSUtil()
+ 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'))
+ record.efficiency = (record.used/(record.total-record.owner))*100
+ record.ts = time - UTC_DIFF
+
+ obj = ObjectUpdate(self.app.model, record, self.cls)
+ self.app.update_thread.enqueue(obj)
+ log.info("PlumageSessionThread--catch-up: pass
completed")
+ sleep(300)
+ log.info("CatchUpPlumageSessionThread--catch-up: catch-up run
completed for records newer than %s and older than %s" % (newest, datetime.now() -
timedelta(seconds=300) + UTC_DIFF))
+ else:
+ log.info("CatchUpPlumageSessionThread: Skipping catch-up, no
records present (probably first-run)")
+ except Exception, e:
+ log.info("%s got exception %s, exiting" % (self.__class__.__name__,
str(e)))