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)))