""" mahali_summary_service.py $Id: mahali_summary_service.py 485 2015-08-14 14:01:39Z flind $ This service periodically collects summary information and places a daily and a long term version into a summary path as json objects. """ import time import os import io import sys import optparse import threading import numpy import json import redis import signal import traceback import daemon # for running as a daemon import ConfigParser # don't normally like to import this way but coded it first and # then moved to library. Should fixup later with better namespace # control... from mahali_common import * class SummaryData: def __init__(self, redis_db, logQ, config, options): self.start_time = time.time() self.redis_db = redis_db self.logQ = logQ self.config = config self.options = options self.fid = None # update time intervals self.update_time = -1 self.short_interval_time = -1 self.extended_interval_time = -1 # tracking data self.summary_times = [] self.uptime = [] self.active_time = [] self.control_state = [] self.storage = [] self.temperature = [] self.temperature_times = [] self.extended_temperature = [] self.extended_temperature_times = [] # data summary dictionary self.data = {} if self.options.path: self.summary_path = self.options.path else: self.summary_path = config['mahali-summary-service']['summary_path'] self.summary_path = fixup_path(self.summary_path) self.device_id = config['mahali-site']['system'] + '-' + config['mahali-site']['name'] + '-' + config['mahali-site']['id'] self.summary_file = self.device_id + '-summary.json' if self.options.debug: self.logQ.logDebug('output path is %s in file %s' % (self.summary_path, self.summary_file)) try: self.__build_path(self.summary_path) except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with summary service path creation for %s." % (exp_str, self.summary_path) self.logQ.logError(emsg) def __build_path(self, fpath): if not os.path.exists(os.path.dirname(fpath)): os.makedirs(os.path.dirname(fpath)) if self.options.debug and self.options.verbose: self.logQ.logDebug('build_path: created %s' % (fpath)) def __open(self, fpath, fname, fmode='w'): self.fid = io.open(fpath+fname,fmode,encoding='UTF-8') if self.options.debug and self.options.verbose: self.logQ.logDebug('open: %s' % (fpath+fname)) def __close(self): if self.fid: self.fid.close() if self.options.debug and self.options.verbose: self.logQ.logDebug('close: closed %s/%s' % (self.summary_path,self.summary_file)) else: if self.options.debug and self.options.verbose: self.logQ.logDebug('close: no file open for %s/%s' % (self.summary_path,self.summary_file)) def loadData(self): if self.options.debug and self.options.verbose: self.logQ.logDebug('loadData') try: self.__open(self.summary_path,self.summary_file,'r') except Exception as eobj: exp_str = str(ExceptionString(eobj)) if self.options.debug: self.logQ.logDebug("exception: %s. Object does not exists for loading in %s%s." % (exp_str, self.summary_path, self.summary_file)) if self.fid: dbuf = self.fid.read() try: jobj = deserialize_json(dbuf) except Exception as eobj: exp_str = str(ExceptionString(eobj)) # this is very common and happens on any box with no initial summary object if self.options.debug: self.logQ.logDebug("exception: %s. Unable to translate object %s in %s%s." % (exp_str, dbuf, self.summary_path, self.summary_file)) self.__close() return if jobj['mahali-site']['system'] == self.config['mahali-site']['system'] and \ jobj['mahali-site']['name'] == self.config['mahali-site']['name'] and \ jobj['mahali-site']['id'] == self.config['mahali-site']['id'] and \ jobj['mahali-site']['install_date'] == self.config['mahali-site']['install_date']: self.data = jobj # now back copy historical data into arrays try: self.summary_times = self.data[self.device_id]['summary_times'] self.uptime = self.data[self.device_id]['uptime'] self.active_time = self.data[self.device_id]['active_time'] self.control_state = self.data[self.device_id]['control_state'] self.storage = self.data[self.device_id]['storage'] self.temperature = self.data[self.device_id]['temperature'] self.temperature_times = self.data[self.device_id]['temperature_times'] self.extended_temperature = self.data[self.device_id]['extended_temperature'] self.extended_temperature_times = self.data[self.device_id]['extended_temperature_times'] except Exception as eobj: exp_str = str(ExceptionString(eobj)) if self.options.debug: self.logQ.logDebug("exception: %s. Problem copying historical data from %s%s." % (exp_str, self.summary_path, self.summary_file)) self.__close() if self.options.debug: self.logQ.logDebug("Loaded historical data from %s/%s." % (self.summary_path, self.summary_file)) else: if self.options.debug: self.logQ.logDebug("Stale historical data in %s/%s, not loaded." % (self.summary_path, self.summary_file)) self.__close() def writeData(self): if self.options.debug and self.options.verbose: self.logQ.logDebug('writeData') try: self.__open(self.summary_path,self.summary_file,'w') except Exception as eobj: exp_str = str(ExceptionString(eobj)) if self.options.debug: self.logQ.logDebug("exception: %s. Unable to create object file %s%s." % (exp_str, self.summary_path, self.summary_file)) return if self.fid: # check we are active if not self.config['mahali-summary-service']['summary_data']: if self.options.debug: self.logQ.logDebug('summary service write attempt with symmary data flag disabled') self.__close() return # serialize for output try: dbuf = serialize_json(self.data) except Exception as eobj: exp_str = str(ExceptionString(eobj)) if self.options.debug: self.logQ.logDebug("exception: %s. Unable to translate data for output in %s%s." % (exp_str, dbuf, self.summary_path, self.summary_file)) self.__close() return # write the output self.fid.write(unicode(dbuf + '\n')) self.fid.flush() if self.options.debug and self.options.verbose: self.logQ.logDebug('log: data written and flushed for %s/%s' % (self.summary_path,self.summary_file)) # close output self.__close() def mirrorData(self): if self.options.debug and self.options.verbose: self.logQ.logDebug('mirrorData') active_time = time.time() - self.start_time state_info = {'start_time':self.start_time, 'active_time':active_time, 'update_time':self.update_time} # watch out for recursion on including the "data" field mahali_set_object(self.redis_db,self.logQ,'mahali-summary-service-state', state_info) def updateData(self): if self.options.debug and self.options.verbose: self.logQ.logDebug('updateData') self.update_time = time.time() # update timestamp self.summary_times.append(self.update_time) # update static configuration information # this is a copy of the configuraiton information # the update service gets most items #for k in self.config.keys(): # cfg = self.config[k] try: self.data.update(self.config) except Exception as eobj: exp_str = str(ExceptionString(eobj)) if self.options.debug: self.logQ.logDebug("exception: %s. Unable to update config data." % (exp_str)) # update current state from object mirrors # grab each service, read state from mirror, update for k in self.config.keys(): #print "key is %s" %(k) if 'service' in k: s_obj = {} try: #print '%s-state' % (k) s_state = mahali_get_object(self.redis_db, self.logQ, '%s-state' %(k)) if s_state: s_obj = {'%s-state' % (k) : s_state} else: s_obj = None except Exception as eobj: exp_str = str(ExceptionString(eobj)) if self.options.debug: # this is often just it not existing in redis self.logQ.logDebug("Update unable to get state for service %s." % (exp_str, k)) try: # a None object means the data does not exist in redis, only update good data if s_obj: self.data.update(s_obj) except Exception as eobj: exp_str = str(ExceptionString(eobj)) if self.options.debug: self.logQ.logDebug("exception: %s. Unable to copy state to data for service %s." % (exp_str, k)) # historical data includes : temperature, uptime, active time, storage space # this could all be more elegant and generic. Something like a dataframe pattern # update device current state parameters uptime = get_uptime() storage = get_storage() self.uptime.append(uptime) self.storage.append(storage) # grab active time from controller state which was previously mirrored self.active_time.append(self.data['mahali-control-service-state']['active_time']) self.control_state.append(self.data['mahali-control-service-state']['control_state']) summary_interval = self.config['mahali-summary-service']['summary_interval'] short_extent = self.config['mahali-summary-service']['summary_short_extent'] long_extent = self.config['mahali-summary-service']['summary_long_extent'] extended_interval = int(summary_interval * float(long_extent)/float(short_extent)) # save historical short term data if self.summary_times[-1] - self.short_interval_time > summary_interval: self.short_interval_time = self.summary_times[-1] temperature = self.data['mahali-relay-service-state']['temperature'] temperature_time = self.data['mahali-relay-service-state']['temperature_time'] self.temperature.append(temperature) self.temperature_times.append(temperature_time) # save historical long term data # pure decimation (no averaging, etc) for the extended interval if self.summary_times[-1] - self.extended_interval_time > extended_interval: self.extended_interval_time = self.summary_times[-1] temperature = self.data['mahali-relay-service-state']['temperature'] temperature_time = self.data['mahali-relay-service-state']['temperature_time'] self.extended_temperature.append(temperature) self.extended_temperature_times.append(temperature_time) print "device id: %s" %(self.device_id) # build device state device_state = {self.device_id:{'summary_times':self.summary_times, 'uptime':self.uptime, 'active_time':self.active_time, 'storage':self.storage, 'control_state':self.control_state, 'temperature':self.temperature, 'temperature_times':self.temperature_times, 'extended_temperature':self.extended_temperature, 'extended_temperature_times':self.extended_temperature_times}} # update device state dictionary self.data.update(device_state) def __getExpireList(self, times, now, limit): idx = 0 expire_list = [] for st in times: if now - st > limit: expire_list.append(idx) idx += 1 return expire_list def expireData(self): if self.options.debug and self.options.verbose: self.logQ.logDebug('expireData') now = time.time() # time limits short_extent = self.config['mahali-summary-service']['summary_short_extent'] long_extent = self.config['mahali-summary-service']['summary_long_extent'] # get decisions short_del_list = self.__getExpireList(self.data[self.device_id]['summary_times'], now, short_extent) # temperatures short_temp_del_list = self.__getExpireList(self.data[self.device_id]['temperature_times'], now, short_extent) # extended temperatures long_del_list = self.__getExpireList(self.data[self.device_id]['extended_temperature_times'], now, long_extent) # debug if self.options.debug and self.options.verbose: self.logQ.logDebug('summary service expire lists') self.logQ.logDebug(short_del_list) self.logQ.logDebug(short_temp_del_list) self.logQ.logDebug(long_del_list) # now delete everything flagged if short_del_list: for d in short_del_list: if d in range(len(self.data[self.device_id]['summary_times'])): del self.data[self.device_id]['summary_times'][d] del self.data[self.device_id]['uptime'][d] del self.data[self.device_id]['active_time'][d] del self.data[self.device_id]['storage'][d] if short_temp_del_list: for d in short_temp_del_list: if d in range(len(self.data[self.device_id]['temperature_times'])): del self.data[self.device_id]['temperature_times'][d] del self.data[self.device_id]['temperature'][d] if long_del_list: for d in long_del_list: if d in range(len(self.data[self.device_id]['extended_temperature_times'])): del self.data[self.device_id]['extended_temperature_times'][d] del self.data[self.device_id]['extended_temperature'][d] def parse_command_line(): parser = optparse.OptionParser() parser.add_option("-v", "--verbose",action="store_true", dest="verbose", default=False,help="Print status messages to stdout.") parser.add_option("-d", "--debug",action="store_true", dest="debug", default=False,help="Print debug messages to stdout.") parser.add_option("-c", "--config",dest="CPATH",help="Use configuration in path CPATH.") parser.add_option("-s", "--simulate",action="store_true",dest="simulate",default=False,help="Activate in simulator mode.") parser.add_option("-p", "--path",dest="path",help="Output data to provided .") parser.add_option("-f", "--foreground",action="store_true",dest="foreground",help="Execute in foreground and not daemon context.") (options, args) = parser.parse_args() return (options, args) def mahali_summary_service(options): global flags start_time = time.time() # create redis interface redis_db = redis.StrictRedis(host='localhost', port=6379, db=0) # register with redis in the mahali service list mahali_register_service(redis_db, 'mahali-summary-service') # create logging instance logQ = redisLoggerQueue('summary-service',redis_db, options) # load configuration mahali_cfg = mahaliConfiguration('mahali-gps-array', options.CPATH, redis_db, logQ) config = mahali_cfg.load('mahali-summary-config') service_interval = config['mahali-summary-service']['summary_interval'] # create summary object summary_data = SummaryData(redis_db,logQ,config,options) try: # log startup logQ.logInfo('startup') # load existing summary data summary_data.loadData() # wait for exit and update redis heartbeat 1 Hz while True: if options.debug and options.verbose: logQ.logDebug('top of service loop at ' + str(time.gmtime(time.time()))) # update static summary data try: summary_data.updateData() except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with summary service data update." % (exp_str) logQ.logError(emsg) # expire old summary data try: summary_data.expireData() except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with summary service data expiration." % (exp_str) logQ.logError(emsg) # write out summary object try: summary_data.writeData() except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with summary service data write." % (exp_str) logQ.logError(emsg) # mirror summary object to redis try: summary_data.mirrorData() except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with summary service data mirroring." % (exp_str) logQ.logError(emsg) # push out heartbeat active_time = time.time() - start_time # data is a dictionary with key values corresponding to # iso8601 timestamp, unix long second timestamp (redundant), type of measurement, and data value mahali_send_event(redis_db,logQ,'mahali-summary-heartbeat','heartbeat','active',active_time) # wait until the next service update time time.sleep(service_interval) except KeyboardInterrupt: logQ.logInfo("exiting mahali-summary-service on keyboard interrupt") except Exception as eobj: exp_str = str(ExceptionString(eobj)) emsg = "exception: %s. Problem with mahali-summary-service." % (exp_str) logQ.logError(emsg) # publish shutdown logQ.logInfo('shutdown') if not mahali_unregister_service(redis_db, 'mahali-summary-service'): sys.exit() # deregister with redis redis_db.connection_pool.disconnect() if __name__ == '__main__': # parse command line options options, args = parse_command_line() if (options.foreground): print "Mahali Summary Service in DEBUG mode" mahali_summary_service(options) else: with daemon.DaemonContext(): mahali_summary_service(options)