From e1a1760ce19787585c2f8c570feb168ff646a33e Mon Sep 17 00:00:00 2001 From: Matteo Vicenzi Date: Thu, 8 Aug 2024 21:29:24 -0500 Subject: [PATCH 1/9] bank Justin's archiver work --- sbndqm/Archiver/Archiver.py | 311 +++++++++++++------ sbndqm/Archiver/ArchiverConfig.json | 20 +- sbndqm/Archiver/CreateMonitorMap.py | 79 +++++ sbndqm/Archiver/DumpPostgres.py | 78 +++++ sbndqm/Archiver/NewArchiver.py | 398 +++++++++++++++++++++++++ sbndqm/Archiver/RestartArchiver.sh | 14 + sbndqm/Archiver/ddl/channelMonitor.sql | 12 +- sbndqm/Archiver/ddl/createMonitor.sh | 5 +- sbndqm/Archiver/ddl/monitorMap.sql | 35 +-- sbndqm/Archiver/inspect_channel.py | 55 ++++ sbndqm/Archiver/query_channels.py | 74 +++++ 11 files changed, 949 insertions(+), 132 deletions(-) create mode 100644 sbndqm/Archiver/CreateMonitorMap.py create mode 100644 sbndqm/Archiver/DumpPostgres.py create mode 100644 sbndqm/Archiver/NewArchiver.py create mode 100644 sbndqm/Archiver/RestartArchiver.sh create mode 100644 sbndqm/Archiver/inspect_channel.py create mode 100644 sbndqm/Archiver/query_channels.py diff --git a/sbndqm/Archiver/Archiver.py b/sbndqm/Archiver/Archiver.py index 93c85a5..4faf073 100644 --- a/sbndqm/Archiver/Archiver.py +++ b/sbndqm/Archiver/Archiver.py @@ -11,12 +11,36 @@ from multiprocessing import Process import math -def main(args): +def ConnectRedis(Config, args): + r = redis.Redis(host=Config['redis']['hostname'], port=Config['redis']['port'], password=args.password) + try: + r.ping() + except: + print('Archiver Error: trouble connecting to Redis database on server (%s) port (%i)' % (Config['redis']['hostname'], Config['redis']['port'])) + logging.info('ERROR: trouble connecting to the Redis database on server (%s) port (%i)' % (Config['redis']['hostname'], Config['redis']['port'])) + return None + logging.info('Connection to the Redis database has been successfully established on server (%s) port (%i)' % (Config['redis']['hostname'], Config['redis']['port'])) + return r - signal.signal(signal.SIGINT, signal_handler) +def ConnectPostgreSQL(Config): + with open(Config['postgresql']['passfile'], 'r') as passfile: + pwd = passfile.read().replace('\n', '') + p = psycopg2.connect(host=Config['postgresql']['hostname'], database=Config['postgresql']['database'], user=Config['postgresql']['user'], port=Config['postgresql']['port'], password=pwd) + #host=Config['postgresql']['hostname'], database=Config['postgresql']['database'], user='mueller', port=Config['postgresql']['port']) + + try: + cur = p.cursor() + cur.execute('SELECT 1') + except psycopg2.OperationalError: + print('Archiver Error: trouble connecting to PostgreSQL database on server (%s) port (%i)' % (Config['postgresql']['hostname'], Config['postgresql']['port'])) + logging.info('ERROR: trouble connecting to PostgreSQL database on server (%s) port (%i)' % (Config['postgresql']['hostname'], Config['postgresql']['port'])) + return None, None + logging.info('Connection to the PostgreSQL database has been successfully established on server (%s) port (%i)' % (Config['postgresql']['hostname'], Config['postgresql']['port'])) + + return p, cur - #Load configuration information from the configuration file. - with open('ArchiverConfig.json') as JSONFile: +def LoadConfig(Name): + with open(Name) as JSONFile: Config = json.load(JSONFile) mode = 'a' @@ -26,56 +50,97 @@ def main(args): filemode=mode, format='%(asctime)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S') + return Config + +def main(args): + + signal.signal(signal.SIGINT, signal_handler) + + #Load configuration information from the configuration file. + Config = LoadConfig('ArchiverConfig.json') logging.info('Configuration loaded. Starting Archiver...') #Connect to the Redis database. - r = redis.Redis(host=Config['redis']['hostname'], port=Config['redis']['port'], password=args.password) - try: - r.ping() - except: - print('Archiver Error: trouble connecting to Redis database on server (%s) port (%i)' % (Config['redis']['hostname'], Config['redis']['port'])) - logging.info('ERROR: trouble connecting to the Redis database on server (%s) port (%i)' % (Config['redis']['hostname'], Config['redis']['port'])) - return - logging.info('Connection to the Redis database has been successfully established on server (%s) port (%i)' % (Config['redis']['hostname'], Config['redis']['port'])) + r = ConnectRedis(Config, args) + if r is None: + sys.exit(0) #Connect to the PostgreSQL database. - p = psycopg2.connect(host=Config['postgresql']['hostname'], database=Config['postgresql']['database'], user=Config['postgresql']['user'], port=Config['postgresql']['port']) - try: - cur = p.cursor() - cur.execute('SELECT 1') - except psycopg2.OperationalError: - print('Archiver Error: trouble connecting to PostgreSQL database on server (%s) port (%i)' % (Config['postgresql']['hostname'], Config['postgresql']['port'])) - logging.info('ERROR: trouble connecting to PostgreSQL database on server (%s) port (%i)' % (Config['postgresql']['hostname'], Config['postgresql']['port'])) - return - logging.info('Connection to the PostgreSQL database has been successfully established on server (%s) port (%i)' % (Config['postgresql']['hostname'], Config['postgresql']['port'])) + p, cur = ConnectPostgreSQL(Config) + if cur is None: + sys.exit(0) + #This next bit deletes the record of streams that have been completed. This is for testing purposes only. - for key in r.scan_iter("*_LatestCompleted"): - r.delete(key) + #for key in r.scan_iter('PMT*_LatestCompleted'): + # r.delete(key) + + #for key in r.scan_iter("*_LatestCompleted"): + #print(key) + #if key == 'tpc_channel:1259:rms:archiving': + # print(key[:-16], 'Reseting') + #CurrentTime = int(GetLatest(r,key[:-16]).split('-')[0]) + #print('Current time', CurrentTime) + #SetLatest(r,key[:-16], '1666210000000') + #SetLatest(r,key[:-16], '1713834881000') #Now we split the job into multiple processes that each handle some portion of the streams. - cur.execute('SELECT * FROM ' + Config['postgresql']['metric_config'] + ';') - StreamConfig = cur.fetchall() + try: + cur.execute('SELECT * FROM ' + Config['postgresql']['metric_config'] + ';') + StreamConfig = cur.fetchall() + except Exception as err: + logging.error('Error while querying metric config table: {}'.format(err)) + logging.error(type(err)) + sys.exit(0) ProcessCount = int(args.processes) ProcessList = [] #Start the Monitor process, which monitors CPU usage, memory, and sends a heartbeat to Redis. Process(target=Monitor, args=(r, Config)).start() for i in range(ProcessCount): KeyList = [ x[0] + ':' + str(x[2]) + ':' + x[1] + ':' + 'archiving' for x in StreamConfig ][i::ProcessCount] - ProcessList.append(Process(target=ProcessStreams, args=(r, p, cur, { x : GetLatest(r, x) for x in KeyList }, { x[0] + ':' + str(x[2]) + ':' + x[1] + ':' + 'archiving' : x for x in StreamConfig[i::ProcessCount]}))) + ProcessList.append(Process(target=ProcessStreams, args=(r, p, cur, { x : GetLatest(r, x) for x in KeyList }, { x[0] + ':' + str(x[2]) + ':' + x[1] + ':' + 'archiving' : x for x in StreamConfig[i::ProcessCount]}, args))) for i in range(ProcessCount): ProcessList[i].start() for i in range(ProcessCount): ProcessList[i].join() -def ProcessStreams(r, p, cur, StreamDict, Config): +def ProcessStreams(r, p, cur, StreamDict, Config, args): + MetricDict = { x : [] for x in StreamDict.keys()} logging.info('Archiver in ProcessStreams.') + logging.info('LatestCompleted for ' + StreamDict.keys()[0] + ': ' + StreamDict[StreamDict.keys()[0]]) + #We now begin to continuously query these streams for new metrics, archiving them when possible. while True: + TotalSetTime = 0 + TotalSetN = 0 #ReadStream is a list of stream object which have new entries that have yet to be archived. - ReadStream = r.xread(StreamDict, block=0) + try: + st = time.time() + ReadStream = r.xread(StreamDict, block=0) + logging.info('XREAD time: {}'.format(time.time()-st)) + #logging.info('Redis READ completed.') + except redis.RedisError as err: + logging.error('Error while reading streams: {}'.format(err)) + logging.error(type(err)) + if ReconnectCount == 4: + sys.exit(0) + else: + ReconnectCount += 1 + logging.error('Reconnecting to Redis (Attempt ' + str(ReconnectCount) + ') ...') + ArcConfig = LoadConfig('ArchiverConfig.json') + r = ConnectRedis(Config, args) + continue + + if len(ReadStream) == 0: + logging.error('Block timeout. Reconnecting to Redis...') + ArcConfig = LoadConfig('ArchiverConfig.json') + r = ConnectRedis(Config, args) + continue + + ReconnectCount = 0 + #Loop over the streams which have entries to be archived. for StreamObject in ReadStream: #Loop over the individual entries in the stream. @@ -83,95 +148,112 @@ def ProcessStreams(r, p, cur, StreamDict, Config): #Check if the latest completed archived entry is '0' (no metrics have been archived yet for the stream). if StreamDict[StreamObject[0]] == '0': NewLatest = str( int( DataObject[0].split('-')[0] ) - 1 ) + '-0' - SetLatest(r, StreamObject[0], NewLatest) + TotalSetTime += SetLatest(r, StreamObject[0], NewLatest) + TotalSetN += 1 ProcessData(MetricDict, DataObject, Config, StreamObject[0]) StreamDict[StreamObject[0]] = DataObject[0] #Check if the current entry is within the time block (contained in the interval of entries to be archived). - elif int( DataObject[0].split('-')[0] ) < int( GetLatest(r, StreamObject[0]).split('-')[0] ) + 1000*Config[StreamObject[0]][3]: + elif int( DataObject[0].split('-')[0] ) < int( GetLatest(r, StreamObject[0]).split('-')[0] ) + 1000*Config[StreamObject[0]][5]: + ProcessData(MetricDict, DataObject, Config, StreamObject[0]) + StreamDict[StreamObject[0]] = DataObject[0] + #Make sure that there are metrics to archive (No metrics in MetricDict, but GetLatest is more than one time block previous to the next entry). + elif int( DataObject[0].split('-')[0] ) > int( GetLatest(r, StreamObject[0]).split('-')[0] ) + 1000*Config[StreamObject[0]][5] and len(MetricDict[StreamObject[0]]) == 0: + NewLatest = str( int( DataObject[0].split('-')[0] ) - 1 ) + '-0' + TotalSetTime += SetLatest(r, StreamObject[0], NewLatest) + TotalSetN += 1 ProcessData(MetricDict, DataObject, Config, StreamObject[0]) StreamDict[StreamObject[0]] = DataObject[0] #The entry is outside the current time block. Set a new latest completed and write to the database. else: #Case for "mean" averaging. if Config[StreamObject[0]][3] == 0: - SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + TotalSetN += 1 WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][1], int(MetricDict[StreamObject[0]][0].split('-')[0])) MetricDict[StreamObject[0]] = [] #Case for "median" averaging. elif Config[StreamObject[0]][3] == 1: - SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][-1][0]) + TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][-1][0]) + TotalSetN += 1 MetricList = [ x[1] for x in MetricDict[StreamObject[0]] ] - WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], median(MetricList), int(MetricDict[StreamObject[0]][0].split('-')[0])) + WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], median(MetricList), int(MetricDict[StreamObject[0]][-1][0].split('-')[0])) MetricDict[StreamObject[0]] = [] #Case for "mode" averaging. elif Config[StreamObject[0]][3] == 2: - SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + TotalSetN += 1 WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][3], int(MetricDict[StreamObject[0]][0].split('-')[0])) MetricDict[StreamObject[0]] = [] #Case for "max" averaging. elif Config[StreamObject[0]][3] == 3: - SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + TotalSetN += 1 WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][1], int(MetricDict[StreamObject[0]][0].split('-')[0])) MetricDict[StreamObject[0]] = [] #Case for "min" averaging. elif Config[StreamObject[0]][3] == 4: - SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + TotalSetN += 1 WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][1], int(MetricDict[StreamObject[0]][0].split('-')[0])) MetricDict[StreamObject[0]] = [] #Case for "last" averaging. elif Config[StreamObject[0]][3] == 5: - SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][1], int(MetricDict[StreamObject[0]][0].split('-')[0])) MetricDict[StreamObject[0]] = [] #After performing the archiving on the previous block of data, we can now process the current entry. ProcessData(MetricDict, DataObject, Config, StreamObject[0]) StreamDict[StreamObject[0]] = DataObject[0] #Check to see if there is a gap in the data stream (the current object is more than one time interval from the latest completed). - if int( DataObject[0].split('-')[0] ) > int( GetLatest(r, StreamObject[0]).split('-')[0] ) + 1000*Config[StreamObject[0]][3]: + if int( DataObject[0].split('-')[0] ) > int( GetLatest(r, StreamObject[0]).split('-')[0] ) + 1000*Config[StreamObject[0]][5]: NewLatest = str( int( DataObject[0].split('-')[0] ) - 1 ) + '-0' - SetLatest(r, StreamObject[0], NewLatest) + TotalSetTime += SetLatest(r, StreamObject[0], NewLatest) + TotalSetN += 1 + logging.info('Total set time: {}'.format(TotalSetTime)) + logging.info('Average set time: {}'.format(float(TotalSetTime)/float(TotalSetN) if TotalSetN else 0)) def ProcessData(Metrics, Datum, Config, StreamName): Value = float(ReadDatum(Datum[1])) - if Config[StreamName][3] == 0: - if len(Metrics[StreamName]) == 0: - Metrics[StreamName] = [Datum[0], Value, 1] - else: - Metrics[StreamName][0] = Datum[0] - Metrics[StreamName][1] = (Metrics[StreamName][1] * Metrics[StreamName][2] + Value) / (Metrics[StreamName][2] + 1) - Metrics[StreamName][2] += 1 - elif Config[StreamName][3] == 1: #Keeps all data for now. - Metrics[StreamName].append( (Datum[0], Value) ) - elif Config[StreamName][3] == 2: - if len(Metrics[StreamName]) == 0: - Metrics[StreamName] = [Datum[0], {Value : 1}, 1, Value] - else: - Metrics[StreamName][0] = Datum[0] - if Value not in Metrics[StreamName][1]: - Metrics[StreamName][1][Value] = 1 + if not math.isnan(Value): + if Config[StreamName][3] == 0: + if len(Metrics[StreamName]) == 0: + Metrics[StreamName] = [Datum[0], Value, 1] + else: + Metrics[StreamName][0] = Datum[0] + Metrics[StreamName][1] = (Metrics[StreamName][1] * Metrics[StreamName][2] + Value) / (Metrics[StreamName][2] + 1) + Metrics[StreamName][2] += 1 + elif Config[StreamName][3] == 1: #Keeps all data for now. + Metrics[StreamName].append( (Datum[0], Value) ) + elif Config[StreamName][3] == 2: + if len(Metrics[StreamName]) == 0: + Metrics[StreamName] = [Datum[0], {Value : 1}, 1, Value] else: - Metrics[StreamName][1][Value] += 1 - if Metrics[StreamName][1][Value] > Metrics[StreamName][2]: - Metrics[StreamName][2] = Metrics[StreamName][1][Value] - Metrics[StreamName][3] = Value - elif Config[StreamName][3] == 3: - if len(Metrics[StreamName]) == 0: - Metrics[StreamName] = [Datum[0], Value] - else: - Metrics[StreamName][0] = Datum[0] - if Metrics[StreamName][1] < Value: Metrics[StreamName][1] = Value - elif Config[StreamName][3] == 4: - if len(Metrics[StreamName]) == 0: - Metrics[StreamName] = [Datum[0], Value] - else: - Metrics[StreamName][0] = Datum[0] + Metrics[StreamName][0] = Datum[0] + if Value not in Metrics[StreamName][1]: + Metrics[StreamName][1][Value] = 1 + else: + Metrics[StreamName][1][Value] += 1 + if Metrics[StreamName][1][Value] > Metrics[StreamName][2]: + Metrics[StreamName][2] = Metrics[StreamName][1][Value] + Metrics[StreamName][3] = Value + elif Config[StreamName][3] == 3: + if len(Metrics[StreamName]) == 0: + Metrics[StreamName] = [Datum[0], Value] + else: + Metrics[StreamName][0] = Datum[0] + if Metrics[StreamName][1] < Value: Metrics[StreamName][1] = Value + elif Config[StreamName][3] == 4: + if len(Metrics[StreamName]) == 0: + Metrics[StreamName] = [Datum[0], Value] + else: + Metrics[StreamName][0] = Datum[0] if Metrics[StreamName][1] > Value: Metrics[StreamName][1] = Value - elif Config[StreamName][3] == 5: - if len(Metrics[StreamName]) == 0: - Metrics[StreamName] = [Datum[0], Value] - else: - Metrics[StreamName][0] = Datum[0] - Metrics[StreamName][1] = Value + elif Config[StreamName][3] == 5: + if len(Metrics[StreamName]) == 0: + Metrics[StreamName] = [Datum[0], Value] + else: + Metrics[StreamName][0] = Datum[0] + Metrics[StreamName][1] = Value def TypeToStructType(Name): if Name == "int8_t": return "b" @@ -212,14 +294,26 @@ def ReadDatum(dat): return ParseBinary(val,key)[0] def GetLatest(Database, StreamName): - Latest = Database.get(StreamName + '_LatestCompleted') + try: + Latest = Database.get(StreamName + '_LatestCompleted') + except redis.RedisError as err: + logging.error(str('Error while getting latest for stream ' + StreamName + ': {}').format(err)) + logging.error(type(err)) + sys.exit(0) if Latest is not None: return Latest else: Database.set(StreamName + '_LatestCompleted','0') return '0' def SetLatest(Database, StreamName, Latest): - response = Database.set(StreamName + '_LatestCompleted', Latest) + s = time.time() + try: + Database.set(StreamName + '_LatestCompleted', Latest) + except redis.RedisError as err: + logging.error(str('Error while setting latest for stream ' + StreamName + ' and timestamp ' + Latest + ': {}').format(err)) + logging.error(type(err)) + sys.exit(0) + return time.time() - s def median(l): l = sorted(l) @@ -233,25 +327,58 @@ def median(l): def WritePostgres(p, cur, Table, Channel, Value, Time): Timestamp = int(Time/1000) - Command = 'INSERT INTO ' + Table + ' (CHANNEL_ID, SMPL_TIME, SMPL_VALUE) VALUES (' + str(Channel) + ',to_timestamp(' + str(Timestamp) + '),' + str(Value) + ');' - cur.execute(Command) - p.commit() + Attempt = 0 + if math.isnan(Value): + Value = 0; + Command = 'INSERT INTO runcon_prd.' + Table + ' (CHANNEL_ID, SMPL_TIME, SMPL_VALUE) VALUES (' + str(Channel) + ',to_timestamp(' + str(Timestamp) + '),' + str(Value) + ');' + while Attempt < 5: + try: + Attempt += 1 + cur.execute(Command) + p.commit() + except psycopg2.OperationalError as err: + time.sleep(60) + logging.error(str('Attempt ' + str(Attempt) + ': Possible PostgreSQL connection issue. Re-connecting...')) + Config = LoadConfig('ArchiverConfig.json') + p, cur = ConnectPostgreSQL(Config) + if cur is None: + sys.exit(0) + except Exception as err: + logging.error(str('Attempt ' + str(Attempt) + ': Error while archiving to table ' + Table + ' for channel ' + str(Channel) + ' and value ' + str(Value) + 'with timestamp ' + str(Time) + ': {}').format(err)) + logging.error(type(err)) + time.sleep(60) + #sys.exit(0) + else: break + else: + logging.error('Failed reattempts; exiting...') + sys.exit(0) def Monitor(r, Config): logging.info('Monitor is running.') - Time = time.time() - while True: + ErrorCount = 0 + while ErrorCount < 4: pipeline = r.pipeline() - if time.time() >= Time + 60: + #pipeline.xadd('archiver_heartbeat', {'float': struct.pack('f', 0)}, maxlen=1) + #CPUs = psutil.cpu_percent(interval=Config['redis']['monitor_time'], percpu=True) + #CPUAverage = psutil.cpu_percent(interval=Config['redis']['monitor_time']) + #Mem = psutil.virtual_memory()[2] + #try: + # for i in range(len(CPUs)): pipeline.xadd('archiver_cpu'+str(i), {'float': struct.pack('f', CPUs[i])}, maxlen=10) + # pipeline.xadd('archiver_cpu_average', {'float': struct.pack('f', CPUAverage)}, maxlen=10) + # pipeline.xadd('archiver_mem', {'float': struct.pack('f', Mem)}, maxlen=10) + # [ _ for _ in pipeline.execute()] + #except redis.RedisError as err: + # logging.error(str('Error while executing pipeline for archiver metrics: {}').format(err)) + # sys.exit(0) + #time.sleep(300) + try: pipeline.xadd('archiver_heartbeat', {'float': struct.pack('f', 0)}, maxlen=1) - Time = time.time() - CPUs = psutil.cpu_percent(interval=Config['redis']['monitor_time'], percpu=True) - CPUAverage = psutil.cpu_percent(interval=Config['redis']['monitor_time']) - Mem = psutil.virtual_memory()[2] - for i in range(len(CPUs)): pipeline.xadd('archiver_cpu'+str(i), {'float': struct.pack('f', CPUs[i])}) - pipeline.xadd('archiver_cpu_average', {'float': struct.pack('f', CPUAverage)}) - pipeline.xadd('archiver_mem', {'float': struct.pack('f', Mem)}) - [ _ for _ in pipeline.execute()] + [ _ for _ in pipeline.execute() ] + except redis.RedisError as err: + logging.error(str('Error while executing pipeline for archiver heartbeat: {}').format(err)) + #sys.exit(0) + ErrorCount += 1 + time.sleep(60) def signal_handler(sig, frame): print('Keyboard Interrupt') diff --git a/sbndqm/Archiver/ArchiverConfig.json b/sbndqm/Archiver/ArchiverConfig.json index 7ec9207..40a9776 100644 --- a/sbndqm/Archiver/ArchiverConfig.json +++ b/sbndqm/Archiver/ArchiverConfig.json @@ -1,20 +1,20 @@ { "postgresql": { - "database": "sbnteststand", - "hostname": "cdpgsdev", - "port": 5488, - "user": "mueller", - "metric_config": "example_monitor_map" + "database": "icarus_online_prd", + "hostname": "icarus-db02.fnal.gov", + "port": 5434, + "user": "runcon_writer", + "passfile": "runcon_writer.txt", + "metric_config": "runcon_prd.monitor_map" }, "redis": { - "hostname": "sbnd-daq33", + "hostname": "icarus-db02.fnal.gov", "monitor_time": 3, - "port": 6379, - "timeout": 30 + "port": 6379, "timeout": 30 }, "logging": { - "directory": "/home/nfs/mueller/Archiver/srcs/sbndqm/sbndqm/Archiver", + "directory": "/daq/log/archiver/", "name": "archiver.log", - "overwrite": true + "overwrite": false } } diff --git a/sbndqm/Archiver/CreateMonitorMap.py b/sbndqm/Archiver/CreateMonitorMap.py new file mode 100644 index 0000000..946e11f --- /dev/null +++ b/sbndqm/Archiver/CreateMonitorMap.py @@ -0,0 +1,79 @@ +import psycopg2 + +Database = psycopg2.connect(host='icarus-db01.fnal.gov', database='icarus_online_prd', user='runcon_admin', password='xqr98CEm', port=5434) +Cursor = Database.cursor() + + +########################################## +## DROP And CREATE PMT RMS table +########################################## + +#Cursor.execute('DELETE FROM runcon_prd.pmt_channel_rms_median_monitor;') +#Cursor.execute('CREATE TABLE runcon_prd.pmt_channel_rms_median_monitor (CHANNEL_ID INTEGER NOT NULL, SMPL_TIME TIMESTAMP WITH TIME ZONE NOT NULL, SMPL_VALUE DOUBLE PRECISION, PRIMARY KEY (CHANNEL_ID, SMPL_TIME));') +#Database.commit() + +########################################## +## DROP And CREATE PMT baseline table +########################################## + +#Cursor.execute('DELETE FROM runcon_prd.pmt_channel_baseline_mean_monitor;') +#Cursor.execute('CREATE TABLE runcon_prd.pmt_channel_baseline_mean_monitor (CHANNEL_ID INTEGER NOT NULL, SMPL_TIME TIMESTAMP WITH TIME ZONE NOT NULL, SMPL_VALUE DOUBLE PRECISION, PRIMARY KEY (CHANNEL_ID, SMPL_TIME));') +#Database.commit() + + +########################################## +## Clean (DELETE) runcon_prd.monitor_map +########################################## + +#Cursor.execute('DELETE FROM runcon_prd.monitor_map;') +#Database.commit() + +########################################## +## ADD TPC RMS Metrics to monitoring +########################################## + +#for i in range(0,55296): +# Cursor.execute('INSERT INTO runcon_prd.monitor_map (GROUP_NAME, METRIC, CHANNEL_ID, MONITOR_TYPE, POSTGRES_TABLE, TIME_INTERVAL) VALUES (\'tpc_channel\', \'rms\',' + str(i) + ', 0, \'tpc_channel_rms_mean_monitor\', 1200);') +# Database.commit() + +########################################## +## ADD PMT RMS Metrics to monitoring +########################################## + +#for i in range(1,360): +# Cursor.execute('INSERT INTO runcon_prd.monitor_map (GROUP_NAME, METRIC, CHANNEL_ID, MONITOR_TYPE, POSTGRES_TABLE, TIME_INTERVAL) VALUES (\'PMT\', \'rms\',' + str(i) + ', 1, \'pmt_channel_rms_median_monitor\', 3600);') +# Database.commit() + +########################################## +## ADD PMT baseline Metrics to monitoring +########################################## + +#for i in range(0,360): +# Cursor.execute('INSERT INTO runcon_prd.monitor_map (GROUP_NAME, METRIC, CHANNEL_ID, MONITOR_TYPE, POSTGRES_TABLE, TIME_INTERVAL) VALUES (\'PMT\', \'baseline\',' + str(i) + ', 0, \'pmt_channel_baseline_mean_monitor\', 3600);') +# Database.commit() + +########################################## +## Example fetch and print +########################################## + +#Cursor.execute('SELECT * FROM monitor_map;') +#StreamConfig = Cursor.fetchall() +#for i in range(25): +# print(StreamConfig[i]) + +########################################## +## DROP And CREATE TPC RMS table (OLD) +########################################## + +#Cursor.execute('DROP TABLE tpc_channel_rms_mean_monitor') +#Cursor.execute('CREATE TABLE tpc_channel_rms_mean_monitor (CHANNEL_ID INTEGER NOT NULL, SMPL_TIME TIMESTAMP WITH TIME ZONE NOT NULL, SMPL_VALUE DOUBLE PRECISION, PRIMARY KEY (CHANNEL_ID, SMPL_TIME));') + +########################################## +## Create Monitor Map (OLD) +########################################## + +#Cursor.execute('DROP TABLE monitor_map;') +#Database.commit() + +#Cursor.execute('CREATE TABLE monitor_map (GROUP_NAME CHARACTER VARYING(128) NOT NULL, METRIC CHARACTER VARYING(32) NOT NULL, CHANNEL_ID INTEGER NOT NULL, MONITOR_TYPE INTEGER NOT NULL, POSTGRES_TABLE CHARACTER VARYING(128) NOT NULL, TIME_INTERVAL DOUBLE PRECISION NOT NULL, CREATE_TIME TIMESTAMP WITH TIME ZONE DEFAULT NOW(), PRIMARY KEY (GROUP_NAME,METRIC,CHANNEL_ID,MONITOR_TYPE));') +#Database.commit() diff --git a/sbndqm/Archiver/DumpPostgres.py b/sbndqm/Archiver/DumpPostgres.py new file mode 100644 index 0000000..41b21d3 --- /dev/null +++ b/sbndqm/Archiver/DumpPostgres.py @@ -0,0 +1,78 @@ +import psycopg2 +import datetime as dt +import numpy as np +import pandas as pd +import time + +def Plane(Channel): + PChannel = Channel % 13824 + p = 0 + if PChannel > 2303 and PChannel < 8064: p = 1 + elif PChannel >= 8064: p = 2 + return p + 3*int(Channel / 13824) + +def HalfCrate(Channel): + return int(Channel / 288) + +def main(): + Granularity = 24 + Database = psycopg2.connect(host='icarus-db01.fnal.gov', database='icarus_online_prd', user='mueller', port=5434) + + Today = dt.datetime.today() + for day in range(0, 90): + Lower = Today - dt.timedelta(days=day) + Upper = Lower + dt.timedelta(days=1) + LS = Lower.strftime('%Y-%m-%d') + US = Upper.strftime('%Y-%m-%d') + Cursor = Database.cursor() + #Cursor.execute("SELECT * FROM runcon_prd.tpc_channel_rms_mean_monitor WHERE smpl_time > \'2021-03-" + "{:0>2}".format(day) + "\' AND smpl_time < \'2021-04-" + "{:0>2}".format(1) + "\';") + Cursor.execute("SELECT * FROM runcon_prd.tpc_channel_rms_mean_monitor WHERE smpl_time > \'" + LS + "\' AND smpl_time < \'" + US + "\' AND channel_id > 4288 AND channel_id < 4319;") + Entries = Cursor.fetchall() + Cursor.close() + print(len(Entries)) + Data = {'channel': [], 'date': [], 'rms': [], 'halfcrate': []} # 'plane': []} + for i in range(len(Entries)): + Data['channel'].append(Entries[i][0]) + Data['date'].append(Entries[i][1]) + Data['rms'].append(Entries[i][2]) + #Data['plane'].append(Plane(Entries[i][0])) + Data['halfcrate'].append(HalfCrate(Entries[i][0])) + dataframe = pd.DataFrame(Data) + dataframe = FilterZeros(dataframe) + + CompressedDataFrame = pd.DataFrame() + for Date in np.unique([dt.datetime(int(x.strftime('%Y')), int(x.strftime('%m')), int(x.strftime('%d'))) for x in dataframe.date.to_numpy()]): + Time = time.mktime(Date.timetuple()) + 60*30*(24.0/Granularity) + DateCut = SelectByDay(Date, dataframe) + for Hour in range(Granularity): + print('Working on hour: ' + str(Hour) + '...') + BinCut = SelectByHour(Hour, DateCut, Difference=(24.0/Granularity)) + #TempDF = BinCut.groupby('channel').mean().reset_index() + #TempDF = BinCut.groupby('plane').mean().reset_index() + TempDF = BinCut.groupby('halfcrate').mean().reset_index() + TempDF['date'] = Time + Hour*(60 * (24.0/Granularity))*60 + CompressedDataFrame = CompressedDataFrame.append(TempDF) + + CompressedDataFrame.to_csv('Run0_' + str(day) + '.csv', index=False) + #dataframe.to_csv('dump.csv') + +def SelectByDay(Date, Data): + print(Date) + AsArray = Data.date.to_numpy() + Cut = Data.loc[[(x.strftime('%d') == Date.strftime('%d') and x.strftime('%m') == Date.strftime('%m') and x.strftime('%Y') == Date.strftime('%Y')) for x in AsArray]] + return Cut + + +def SelectByHour(Hour, Data, Difference=1): + LowHour = Hour*Difference + HighHour = (Hour+1)*Difference + AsArray = Data.date.to_numpy() + Cut = Data.loc[[(int(x.strftime('%-H')) + float(x.strftime('%M'))/60 < HighHour and int(x.strftime('%-H')) + float(x.strftime('%M'))/60 >= LowHour) for x in AsArray]] + return Cut + +def FilterZeros(Data): + Cut = Data.loc[Data.rms != 0] + return Cut + +if __name__ == '__main__': + main() diff --git a/sbndqm/Archiver/NewArchiver.py b/sbndqm/Archiver/NewArchiver.py new file mode 100644 index 0000000..a7535be --- /dev/null +++ b/sbndqm/Archiver/NewArchiver.py @@ -0,0 +1,398 @@ +import argparse +import redis as rd +import json +import time +import signal +import sys +import psutil +import struct +import psycopg2 +import logging +from multiprocessing import Process +import math + +def connect_redis(cfg, args): + redis = rd.Redis(host=cfg['redis']['hostname'], + port=cfg['redis']['port'], + password=args.password) + try: + redis.ping() + except: + logging.info('ERROR: trouble connecting to the Redis database on server (%s) port (%i)' % (cfg['redis']['hostname'])) + +def ConnectRedis(Config, args): + r = redis.Redis(host=Config['redis']['hostname'], port=Config['redis']['port'], password=args.password) + try: + r.ping() + except: + print('Archiver Error: trouble connecting to Redis database on server (%s) port (%i)' % (Config['redis']['hostname'], Config['redis']['port'])) + logging.info('ERROR: trouble connecting to the Redis database on server (%s) port (%i)' % (Config['redis']['hostname'], Config['redis']['port'])) + return None + logging.info('Connection to the Redis database has been successfully established on server (%s) port (%i)' % (Config['redis']['hostname'], Config['redis']['port'])) + return r + +def ConnectPostgreSQL(Config): + with open(Config['postgresql']['passfile'], 'r') as passfile: + pwd = passfile.read().replace('\n', '') + p = psycopg2.connect(host=Config['postgresql']['hostname'], database=Config['postgresql']['database'], user=Config['postgresql']['user'], port=Config['postgresql']['port'], password=pwd) + + try: + cur = p.cursor() + cur.execute('SELECT 1') + except psycopg2.OperationalError: + print('Archiver Error: trouble connecting to PostgreSQL database on server (%s) port (%i)' % (Config['postgresql']['hostname'], Config['postgresql']['port'])) + logging.info('ERROR: trouble connecting to PostgreSQL database on server (%s) port (%i)' % (Config['postgresql']['hostname'], Config['postgresql']['port'])) + return None, None + logging.info('Connection to the PostgreSQL database has been successfully established on server (%s) port (%i)' % (Config['postgresql']['hostname'], Config['postgresql']['port'])) + + return p, cur + +def LoadConfig(Name): + with open(Name) as JSONFile: + Config = json.load(JSONFile) + + mode = 'a' + if(Config['logging']['overwrite'] == True): mode = 'w' + logging.basicConfig(filename=Config['logging']['directory'] + Config['logging']['name'], + level=logging.INFO, + filemode=mode, + format='%(asctime)s - %(message)s', + datefmt='%Y-%m-%d %H:%M:%S') + return Config + +def main(args): + + signal.signal(signal.SIGINT, signal_handler) + + #Load configuration information from the configuration file. + Config = LoadConfig('ArchiverConfig.json') + logging.info('Configuration loaded. Starting Archiver...') + + #Connect to the Redis database. + r = ConnectRedis(Config, args) + if r is None: + sys.exit(0) + + #Connect to the PostgreSQL database. + p, cur = ConnectPostgreSQL(Config) + if cur is None: + sys.exit(0) + + + #This next bit deletes the record of streams that have been completed. This is for testing purposes only. + #for key in r.scan_iter('PMT*_LatestCompleted'): + # r.delete(key) + + #for key in r.scan_iter("*_LatestCompleted"): + #print(key) + #if key == 'tpc_channel:1259:rms:archiving': + # print(key[:-16], 'Reseting') + #CurrentTime = int(GetLatest(r,key[:-16]).split('-')[0]) + #print('Current time', CurrentTime) + #SetLatest(r,key[:-16], '1596208000000') + #SetLatest(r,key[:-16], '1608237252732') + #SetLatest(r,key[:-16], '1648073157000') + + #Now we split the job into multiple processes that each handle some portion of the streams. + try: + cur.execute('SELECT * FROM ' + Config['postgresql']['metric_config'] + ';') + StreamConfig = cur.fetchall() + except Exception as err: + logging.error('Error while querying metric config table: {}'.format(err)) + logging.error(type(err)) + sys.exit(0) + ProcessCount = int(args.processes) + ProcessList = [] + #Start the Monitor process, which monitors CPU usage, memory, and sends a heartbeat to Redis. + Process(target=Monitor, args=(r, Config)).start() + for i in range(ProcessCount): + KeyList = [ x[0] + ':' + str(x[2]) + ':' + x[1] + ':' + 'archiving' for x in StreamConfig ][i::ProcessCount] + ProcessList.append(Process(target=ProcessStreams, args=(r, p, cur, { x : GetLatest(r, x) for x in KeyList }, { x[0] + ':' + str(x[2]) + ':' + x[1] + ':' + 'archiving' : x for x in StreamConfig[i::ProcessCount]}, args))) + for i in range(ProcessCount): + ProcessList[i].start() + for i in range(ProcessCount): + ProcessList[i].join() + +def ProcessStreams(r, p, cur, StreamDict, Config, args): + + MetricDict = {x : [] for x in StreamDict.keys()} + + logging.info('Archiver is ready to process Redis streams..') + logging.info('Latest completed for ' + StreamDict.keys()[0] + ': ' + StreamDict[StreamDict.keys()[0]]) + + while True: + TotalSetTime = 0 + TotalSetN = 0 + try: + st = time.time() + ReadStream = r.xread(StreamDict, block=0) + logging.info('XREAD time: {}'.format(time.time()-st)) + except redis.RedisError as err: + logging.error('Error while reading streams: {}'.format(err)) + logging.error(type(err)) + if ReconnectCount == 4: + sys.exit(0) + else: + ReconnectCount += 1 + logging.error('Reconnecting to Redis (Attempt ' + str(ReconnectCount) + ') ...') + ArcConfig = LoadConfig('ArchiverConfig.json') + r = ConnectRedis(Config, args) + continue + + if len(ReadStream) == 0: + logging.error('Block timeout. Reconnecting to Redis...') + ArcConfig = LoadConfig('ArchiverConfig.json') + r = ConnectRedis(Config, args) + continue + + ReconnectCount = 0 + + #Loop over the streams which have entries to be archived. + for StreamObject in ReadStream: + #Loop over the individual entries in the stream. + for DataObject in StreamObject[1]: + #Check if the latest completed archived entry is '0' (no metrics have been archived yet for the stream). + if StreamDict[StreamObject[0]] == '0': + NewLatest = str( int( DataObject[0].split('-')[0] ) - 1 ) + '-0' + TotalSetTime += SetLatest(r, StreamObject[0], NewLatest) + TotalSetN += 1 + ProcessData(MetricDict, DataObject, Config, StreamObject[0]) + StreamDict[StreamObject[0]] = DataObject[0] + #Check if the current entry is within the time block (contained in the interval of entries to be archived). + elif int( DataObject[0].split('-')[0] ) < int( GetLatest(r, StreamObject[0]).split('-')[0] ) + 1000*Config[StreamObject[0]][5]: + ProcessData(MetricDict, DataObject, Config, StreamObject[0]) + StreamDict[StreamObject[0]] = DataObject[0] + #Make sure that there are metrics to archive (No metrics in MetricDict, but GetLatest is more than one time block previous to the next entry). + elif int( DataObject[0].split('-')[0] ) > int( GetLatest(r, StreamObject[0]).split('-')[0] ) + 1000*Config[StreamObject[0]][5] and len(MetricDict[StreamObject[0]]) == 0: + NewLatest = str( int( DataObject[0].split('-')[0] ) - 1 ) + '-0' + TotalSetTime += SetLatest(r, StreamObject[0], NewLatest) + TotalSetN += 1 + ProcessData(MetricDict, DataObject, Config, StreamObject[0]) + StreamDict[StreamObject[0]] = DataObject[0] + #The entry is outside the current time block. Set a new latest completed and write to the database. + else: + #Case for "mean" averaging. + if Config[StreamObject[0]][3] == 0: + TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + TotalSetN += 1 + WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][1], int(MetricDict[StreamObject[0]][0].split('-')[0])) + MetricDict[StreamObject[0]] = [] + #Case for "median" averaging. + elif Config[StreamObject[0]][3] == 1: + TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][-1][0]) + TotalSetN += 1 + MetricList = [ x[1] for x in MetricDict[StreamObject[0]] ] + WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], median(MetricList), int(MetricDict[StreamObject[0]][-1][0].split('-')[0])) + MetricDict[StreamObject[0]] = [] + #Case for "mode" averaging. + elif Config[StreamObject[0]][3] == 2: + TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + TotalSetN += 1 + WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][3], int(MetricDict[StreamObject[0]][0].split('-')[0])) + MetricDict[StreamObject[0]] = [] + #Case for "max" averaging. + elif Config[StreamObject[0]][3] == 3: + TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + TotalSetN += 1 + WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][1], int(MetricDict[StreamObject[0]][0].split('-')[0])) + MetricDict[StreamObject[0]] = [] + #Case for "min" averaging. + elif Config[StreamObject[0]][3] == 4: + TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + TotalSetN += 1 + WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][1], int(MetricDict[StreamObject[0]][0].split('-')[0])) + MetricDict[StreamObject[0]] = [] + #Case for "last" averaging. + elif Config[StreamObject[0]][3] == 5: + TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][1], int(MetricDict[StreamObject[0]][0].split('-')[0])) + MetricDict[StreamObject[0]] = [] + #After performing the archiving on the previous block of data, we can now process the current entry. + ProcessData(MetricDict, DataObject, Config, StreamObject[0]) + StreamDict[StreamObject[0]] = DataObject[0] + #Check to see if there is a gap in the data stream (the current object is more than one time interval from the latest completed). + if int( DataObject[0].split('-')[0] ) > int( GetLatest(r, StreamObject[0]).split('-')[0] ) + 1000*Config[StreamObject[0]][5]: + NewLatest = str( int( DataObject[0].split('-')[0] ) - 1 ) + '-0' + TotalSetTime += SetLatest(r, StreamObject[0], NewLatest) + TotalSetN += 1 + logging.info('Total set time: {}'.format(TotalSetTime)) + logging.info('Average set time: {}'.format(float(TotalSetTime)/float(TotalSetN))) + +def ProcessData(Metrics, Datum, Config, StreamName): + Value = float(ReadDatum(Datum[1])) + if not math.isnan(Value): + if Config[StreamName][3] == 0: + if len(Metrics[StreamName]) == 0: + Metrics[StreamName] = [Datum[0], Value, 1] + else: + Metrics[StreamName][0] = Datum[0] + Metrics[StreamName][1] = (Metrics[StreamName][1] * Metrics[StreamName][2] + Value) / (Metrics[StreamName][2] + 1) + Metrics[StreamName][2] += 1 + elif Config[StreamName][3] == 1: #Keeps all data for now. + Metrics[StreamName].append( (Datum[0], Value) ) + elif Config[StreamName][3] == 2: + if len(Metrics[StreamName]) == 0: + Metrics[StreamName] = [Datum[0], {Value : 1}, 1, Value] + else: + Metrics[StreamName][0] = Datum[0] + if Value not in Metrics[StreamName][1]: + Metrics[StreamName][1][Value] = 1 + else: + Metrics[StreamName][1][Value] += 1 + if Metrics[StreamName][1][Value] > Metrics[StreamName][2]: + Metrics[StreamName][2] = Metrics[StreamName][1][Value] + Metrics[StreamName][3] = Value + elif Config[StreamName][3] == 3: + if len(Metrics[StreamName]) == 0: + Metrics[StreamName] = [Datum[0], Value] + else: + Metrics[StreamName][0] = Datum[0] + if Metrics[StreamName][1] < Value: Metrics[StreamName][1] = Value + elif Config[StreamName][3] == 4: + if len(Metrics[StreamName]) == 0: + Metrics[StreamName] = [Datum[0], Value] + else: + Metrics[StreamName][0] = Datum[0] + if Metrics[StreamName][1] > Value: Metrics[StreamName][1] = Value + elif Config[StreamName][3] == 5: + if len(Metrics[StreamName]) == 0: + Metrics[StreamName] = [Datum[0], Value] + else: + Metrics[StreamName][0] = Datum[0] + Metrics[StreamName][1] = Value + +def TypeToStructType(Name): + if Name == "int8_t": return "b" + if Name == "int16_t": return "h" + if Name == "int32_t": return "i" + if Name == "int64_t": return "q" + if Name == "uint8_t": return "B" + if Name == "uint16_t": return "H" + if Name == "uint32_t": return "I" + if Name == "uint64_t": return "Q" + if Name == "float": return "f" + if Name == "double": return "d" + +def TypeToSize(Name): + if Name == "int8_t": return 1 + if Name == "int16_t": return 2 + if Name == "int32_t": return 4 + if Name == "int64_t": return 8 + if Name == "uint8_t": return 1 + if Name == "uint16_t": return 2 + if Name == "uint32_t": return 4 + if Name == "uint64_t": return 8 + if Name == "float": return 4 + if Name == "double": return 8 + +def ParseBinary(Binary, TypeName): + Size = TypeToSize(TypeName) + Form = TypeToStructType(TypeName) + ret = [] + for i in range(len(Binary) / Size): + dat = Binary[i*Size : (i+1)*Size] + ret.append(struct.unpack(Form, dat)[0]) + return ret + +def ReadDatum(dat): + for key, val in dat.items(): + if key == 'val' or key == 'dat': return val + return ParseBinary(val,key)[0] + +def GetLatest(Database, StreamName): + try: + Latest = Database.get(StreamName + '_LatestCompleted') + except redis.RedisError as err: + logging.error(str('Error while getting latest for stream ' + StreamName + ': {}').format(err)) + logging.error(type(err)) + sys.exit(0) + if Latest is not None: return Latest + else: + Database.set(StreamName + '_LatestCompleted','0') + return '0' + +def SetLatest(Database, StreamName, Latest): + s = time.time() + try: + Database.set(StreamName + '_LatestCompleted', Latest) + except redis.RedisError as err: + logging.error(str('Error while setting latest for stream ' + StreamName + ' and timestamp ' + Latest + ': {}').format(err)) + logging.error(type(err)) + sys.exit(0) + return time.time() - s + +def median(l): + l = sorted(l) + Length = len(l) + if Length < 1: + return None + if Length % 2 == 0: + return ( l[(Length-1)/2] + l[(Length+1)/2] ) / 2.0 + else: + return l[(Length-1)/2] + +def WritePostgres(p, cur, Table, Channel, Value, Time): + Timestamp = int(Time/1000) + Attempt = 0 + if math.isnan(Value): + Value = 0; + Command = 'INSERT INTO runcon_prd.' + Table + ' (CHANNEL_ID, SMPL_TIME, SMPL_VALUE) VALUES (' + str(Channel) + ',to_timestamp(' + str(Timestamp) + '),' + str(Value) + ');' + while Attempt < 5: + try: + Attempt += 1 + cur.execute(Command) + p.commit() + except psycopg2.OperationalError as err: + time.sleep(60) + logging.error(str('Attempt ' + str(Attempt) + ': Possible PostgreSQL connection issue. Re-connecting...')) + Config = LoadConfig('ArchiverConfig.json') + p, cur = ConnectPostgreSQL(Config) + if cur is None: + sys.exit(0) + except Exception as err: + logging.error(str('Attempt ' + str(Attempt) + ': Error while archiving to table ' + Table + ' for channel ' + str(Channel) + ' and value ' + str(Value) + 'with timestamp ' + str(Time) + ': {}').format(err)) + logging.error(type(err)) + time.sleep(60) + #sys.exit(0) + else: break + else: + logging.error('Failed reattempts; exiting...') + sys.exit(0) + +def Monitor(r, Config): + logging.info('Monitor is running.') + ErrorCount = 0 + while ErrorCount < 4: + pipeline = r.pipeline() + #pipeline.xadd('archiver_heartbeat', {'float': struct.pack('f', 0)}, maxlen=1) + #CPUs = psutil.cpu_percent(interval=Config['redis']['monitor_time'], percpu=True) + #CPUAverage = psutil.cpu_percent(interval=Config['redis']['monitor_time']) + #Mem = psutil.virtual_memory()[2] + #try: + # for i in range(len(CPUs)): pipeline.xadd('archiver_cpu'+str(i), {'float': struct.pack('f', CPUs[i])}, maxlen=10) + # pipeline.xadd('archiver_cpu_average', {'float': struct.pack('f', CPUAverage)}, maxlen=10) + # pipeline.xadd('archiver_mem', {'float': struct.pack('f', Mem)}, maxlen=10) + # [ _ for _ in pipeline.execute()] + #except redis.RedisError as err: + # logging.error(str('Error while executing pipeline for archiver metrics: {}').format(err)) + # sys.exit(0) + #time.sleep(300) + try: + pipeline.xadd('archiver_heartbeat', {'float': struct.pack('f', 0)}, maxlen=1) + [ _ for _ in pipeline.execute() ] + except redis.RedisError as err: + logging.error(str('Error while executing pipeline for archiver heartbeat: {}').format(err)) + #sys.exit(0) + ErrorCount += 1 + time.sleep(60) + +def signal_handler(sig, frame): + print('Keyboard Interrupt') + logging.info('Keyboard interrupt. Exiting Archiver.') + sys.exit(0) + +if __name__ == "__main__": + args = argparse.ArgumentParser() + args.add_argument("-pw", "--password", default=None) + args.add_argument("-pr", "--processes", default=1) + main(args.parse_args()) diff --git a/sbndqm/Archiver/RestartArchiver.sh b/sbndqm/Archiver/RestartArchiver.sh new file mode 100644 index 0000000..8f2be42 --- /dev/null +++ b/sbndqm/Archiver/RestartArchiver.sh @@ -0,0 +1,14 @@ +#!/bin/bash + +echo "Remaining OnMon Archiver.py processes:" +ps aux | grep '[p]ython Archiver.py' + +echo "Killing remaining OnMon Archiver.py processes..." +kill -9 $(ps aux | grep '[p]ython Archiver.py' | awk '{print $2}') + +echo "Restarting OnMon Archiver." +cd /home/nfs/icarus/Archiver/sbndqm/sbndqm/Archiver +source env/bin/activate +python Archiver.py -pr 1 & + +echo "OnMon Archiver restarted." diff --git a/sbndqm/Archiver/ddl/channelMonitor.sql b/sbndqm/Archiver/ddl/channelMonitor.sql index 9a62f43..6b890da 100644 --- a/sbndqm/Archiver/ddl/channelMonitor.sql +++ b/sbndqm/Archiver/ddl/channelMonitor.sql @@ -5,7 +5,7 @@ \echo :CH1 \echo :CHN -CREATE OR REPLACE FUNCTION RUNCON_PRD.MONMON(GROUP_NAME CHARACTER, +CREATE OR REPLACE FUNCTION MONMON(GROUP_NAME CHARACTER, METRIC CHARACTER, MTYPE INTEGER, CH1 INTEGER, @@ -17,21 +17,21 @@ $BODY$ TABLE_NAME CHARACTER VARYING(128); CTR INTEGER; BEGIN - MONITOR_NAMED := (SELECT TYPE_NAME FROM RUNCON_PRD.MONITOR_TYPES WHERE MONITOR_TYPE=MTYPE); + MONITOR_NAMED := (SELECT TYPE_NAME FROM MONITOR_TYPES WHERE MONITOR_TYPE=MTYPE); TABLE_NAME := LOWER(GROUP_NAME || '_' || METRIC || '_' || MONITOR_NAMED || '_MONITOR'); RAISE NOTICE 'Table Name: %',TABLE_NAME; EXECUTE FORMAT(' - CREATE TABLE RUNCON_PRD.%I + CREATE TABLE %I ( CHANNEL_ID INTEGER NOT NULL, SMPL_TIME TIMESTAMP WITH TIME ZONE NOT NULL, SMPL_VALUE DOUBLE PRECISION, - PRIMARY KEY (CHANNEL_ID,SMPL_TIME) + PRIMARY KEY (CHANNEL_ID,SMPL_VALUE) )',TABLE_NAME); CTR := CH1; WHILE CTR <= CHN LOOP EXECUTE FORMAT(' - INSERT INTO RUNCON_PRD.MONITOR_MAP VALUES (%L,%L,%L,%L,%L,3600,NOW()) + INSERT INTO MONITOR_MAP VALUES (%L,%L,%L,%L,%L,3600,NOW()) ',GROUP_NAME,METRIC,CTR,MTYPE,TABLE_NAME); CTR := CTR + 1; END LOOP; @@ -40,5 +40,5 @@ $BODY$ $BODY$ LANGUAGE plpgsql; -SELECT RUNCON_PRD.MONMON(:GROUP_NAME,:METRIC,:MONITOR_TYPE,:CH1,:CHN); +SELECT MONMON(:GROUP_NAME,:METRIC,:MONITOR_TYPE,:CH1,:CHN); diff --git a/sbndqm/Archiver/ddl/createMonitor.sh b/sbndqm/Archiver/ddl/createMonitor.sh index 14cbbf9..d069ac0 100755 --- a/sbndqm/Archiver/ddl/createMonitor.sh +++ b/sbndqm/Archiver/ddl/createMonitor.sh @@ -12,10 +12,7 @@ if [ $# -ne 5 ]; then exit fi -# -# Modify the following psql parameters to suit your own datbase -# -psql -U runcon_admin -h icarus-db01 -p 5434 -d icarus_online_prd \ +psql -U badgett -h cdpgsdev -p 5488 -d sbnteststand \ --set=GROUP_NAME=\'$1\' --set=METRIC=\'$2\' --set=MONITOR_TYPE=$3 \ --set=CH1=$4 --set=CHN=$5 -c "\i channelMonitor.sql" diff --git a/sbndqm/Archiver/ddl/monitorMap.sql b/sbndqm/Archiver/ddl/monitorMap.sql index fd5ab8e..222d65c 100644 --- a/sbndqm/Archiver/ddl/monitorMap.sql +++ b/sbndqm/Archiver/ddl/monitorMap.sql @@ -1,32 +1,27 @@ -CREATE TABLE RUNCON_PRD.MONITOR_TYPES +CREATE TABLE MONITOR_TYPES ( MONITOR_TYPE INTEGER NOT NULL, TYPE_NAME CHARACTER VARYING(32) NOT NULL, PRIMARY KEY (MONITOR_TYPE) ); -INSERT INTO RUNCON_PRD.MONITOR_TYPES VALUES (0,'MEAN'); -INSERT INTO RUNCON_PRD.MONITOR_TYPES VALUES (1,'MEDIAN'); -INSERT INTO RUNCON_PRD.MONITOR_TYPES VALUES (2,'MODE'); -INSERT INTO RUNCON_PRD.MONITOR_TYPES VALUES (3,'MAX'); -INSERT INTO RUNCON_PRD.MONITOR_TYPES VALUES (4,'MIN'); -INSERT INTO RUNCON_PRD.MONITOR_TYPES VALUES (5,'LAST'); +INSERT INTO MONITOR_TYPES VALUES (0,'MEAN'); +INSERT INTO MONITOR_TYPES VALUES (1,'MEDIAN'); +INSERT INTO MONITOR_TYPES VALUES (2,'MODE'); +INSERT INTO MONITOR_TYPES VALUES (3,'MAX'); +INSERT INTO MONITOR_TYPES VALUES (4,'MIN'); +INSERT INTO MONITOR_TYPES VALUES (5,'LAST'); -CREATE TABLE RUNCON_PRD.MONITOR_MAP +CREATE TABLE MONITOR_MAP ( - GROUP_NAME CHARACTER VARYING(128) NOT NULL, - METRIC CHARACTER VARYING(32) NOT NULL, - CHANNEL_ID INTEGER NOT NULL, - MONITOR_TYPE INTEGER NOT NULL REFERENCES MONITOR_TYPES(MONITOR_TYPE), - POSTGRES_TABLE CHARACTER VARYING(128) NOT NULL, - TIME_INTERVAL DOUBLE PRECISION NOT NULL, - CREATE_TIME TIMESTAMP WITH TIME ZONE DEFAULT NOW(), - LAST_SMPL_TIME_A TIMESTAMP WITH TIME ZONE, - LAST_SMPL_TIME_B TIMESTAMP WITH TIME ZONE, - LAST_SMPL_VALUE_A DOUBLE PRECISION, - LAST_SMPL_VALUE_B DOUBLE PRECISION, - LAST_SMPL CHAR(1), + GROUP_NAME CHARACTER VARYING(128) NOT NULL, + METRIC CHARACTER VARYING(32) NOT NULL, + CHANNEL_ID INTEGER NOT NULL, + MONITOR_TYPE INTEGER NOT NULL REFERENCES MONITOR_TYPES(MONITOR_TYPE), + POSTGRES_TABLE CHARACTER VARYING(128) NOT NULL, + TIME_INTERVAL DOUBLE PRECISION NOT NULL, + CREATE_TIME TIMESTAMP WITH TIME ZONE DEFAULT NOW(), PRIMARY KEY (GROUP_NAME,METRIC,CHANNEL_ID,MONITOR_TYPE) ); diff --git a/sbndqm/Archiver/inspect_channel.py b/sbndqm/Archiver/inspect_channel.py new file mode 100644 index 0000000..ef2fd10 --- /dev/null +++ b/sbndqm/Archiver/inspect_channel.py @@ -0,0 +1,55 @@ +import numpy as np +import psycopg2 +import pandas as pd +import datetime as dt +import time + +def query_channel(channel): + database = psycopg2.connect(host='icarus-db01.fnal.gov', + database='icarus_online_prd', + user='mueller', + port=5434) + cursor = database.cursor() + + today = dt.datetime.today() + lower = (today - dt.timedelta(weeks=208)) + upper = today.strftime('%Y-%m-%d') + lower = lower.strftime('%Y-%m-%d') + + table = "runcon_prd.tpc_channel_rms_mean_monitor" + date = "smpl_time > \'" + lower + "\' AND smpl_time < \'" + upper +"\'" + entries = list() + chan = "channel_id = " + str(channel) + comm = "SELECT * FROM " + table + " WHERE " + date + " AND " + chan + ";" + cursor.execute(comm) + entries.extend(cursor.fetchall()) + cursor.close() + entries = list(zip(*entries)) + data = {'channel': entries[0], + 'time': [int(time.mktime(d.timetuple())) for d in entries[1]], + 'rms': entries[2]} + return pd.DataFrame(data) + + +def main(): + df = query_channel(43518) + df.to_csv('ch43518.csv', index=False) + """ + for w in range(4, 104, 4): + print('Querying month: ' + str(int(w/4)) + '.') + df = query_channel(unstable_channels, weeks=w) + print(len(df)) + if w != 4: + pd.concat([pd.read_csv('unstable.csv'), df], + ignore_index=True).to_csv('unstable.csv', index=False) + else: df.to_csv('unstable.csv', index=False) + df = query_channel(stable_channels, weeks=w) + print(len(df)) + if w != 4: + pd.concat([pd.read_csv('stable.csv'), df], + ignore_index=True).to_csv('stable.csv', index=False) + else: df.to_csv('stable.csv', index=False) + """ + +if __name__ == '__main__': + main() diff --git a/sbndqm/Archiver/query_channels.py b/sbndqm/Archiver/query_channels.py new file mode 100644 index 0000000..5cd9a9a --- /dev/null +++ b/sbndqm/Archiver/query_channels.py @@ -0,0 +1,74 @@ +import numpy as np +import psycopg2 +import pandas as pd +import datetime as dt +import time + +def query_channel(channels, weeks=4): + database = psycopg2.connect(host='icarus-db01.fnal.gov', + database='icarus_online_prd', + user='mueller', + port=5434) + cursor = database.cursor() + + today = dt.datetime.today() + lower = (today - dt.timedelta(weeks=weeks)) + upper = (lower + dt.timedelta(weeks=4)).strftime('%Y-%m-%d') + lower = lower.strftime('%Y-%m-%d') + + table = "runcon_prd.tpc_channel_rms_mean_monitor" + date = "smpl_time > \'" + lower + "\' AND smpl_time < \'" + upper +"\'" + entries = list() + for c in channels: + chan = "channel_id = " + str(c) + comm = "SELECT * FROM " + table + " WHERE " + date + " AND " + chan + ";" + cursor.execute(comm) + entries.extend(cursor.fetchall()) + cursor.close() + entries = list(zip(*entries)) + data = {'channel': entries[0], + 'time': [int(time.mktime(d.timetuple())) for d in entries[1]], + 'rms': entries[2]} + return pd.DataFrame(data) + + +def main(): + unstable_channels = [144,432,720,976,1296,1584,3312,3888,4464,4752, + 5040,5328,5904,6192,7056,7920,9936,10224,10512, + 10800,11088,11376,11664,13104,13392,13648,13968, + 14256,16848,17712,18000,18288,18864,19152,20016, + 20592,21168,21456,21712,21744,23152,24048,24624, + 27504,27792,28080,28368,28624,28912,28944,29200, + 329232,31248,31536,32112,32976,33264,34704,34992, + 35280,36720,38448,38736,41040,41616,41904,43056, + 44208,44496,44784,45360,45616,46224,46512,46800, + 47664,47952,48816,49104,49360,49712,50256,51696, + 54000,55152] + stable_channels = [16,304,592,848,1168,1456,3184,3760,4336,4624,4912, + 5200,5776,6064,6928,7792,9808,10096,10384,10672, + 10960,11248,11536,12976,13264,13520,13840,14128, + 16720,17584,17872,18160,18736,19024,19888,20464, + 21040,21328,21584,21616,23024,23920,24496,27376, + 27664,27952,28240,28496,28784,28816,29072,29104, + 31120,31408,31984,32848,33136,34576,34864,35152, + 36592,38320,38608,40912,41488,41776,42928,44080, + 44368,44656,45232,45488,46096,46384,46672,47536, + 47824,48688,48976,49232,49584,50128,51568,53872, + 55024] + for w in range(4, 104, 4): + print('Querying month: ' + str(int(w/4)) + '.') + df = query_channel(unstable_channels, weeks=w) + print(len(df)) + if w != 4: + pd.concat([pd.read_csv('unstable.csv'), df], + ignore_index=True).to_csv('unstable.csv', index=False) + else: df.to_csv('unstable.csv', index=False) + df = query_channel(stable_channels, weeks=w) + print(len(df)) + if w != 4: + pd.concat([pd.read_csv('stable.csv'), df], + ignore_index=True).to_csv('stable.csv', index=False) + else: df.to_csv('stable.csv', index=False) + +if __name__ == '__main__': + main() From e5a7a851cda3612ae5af4dc035a3b9eefef2d70c Mon Sep 17 00:00:00 2001 From: Matteo Vicenzi Date: Fri, 9 Aug 2024 08:19:37 -0500 Subject: [PATCH 2/9] remove test scripts, fix indentation --- sbndqm/Archiver/ArchiverConfig.json | 3 +- sbndqm/Archiver/CreateMonitorMap.py | 79 ----------------------------- sbndqm/Archiver/DumpPostgres.py | 78 ---------------------------- sbndqm/Archiver/inspect_channel.py | 55 -------------------- sbndqm/Archiver/query_channels.py | 74 --------------------------- 5 files changed, 2 insertions(+), 287 deletions(-) delete mode 100644 sbndqm/Archiver/CreateMonitorMap.py delete mode 100644 sbndqm/Archiver/DumpPostgres.py delete mode 100644 sbndqm/Archiver/inspect_channel.py delete mode 100644 sbndqm/Archiver/query_channels.py diff --git a/sbndqm/Archiver/ArchiverConfig.json b/sbndqm/Archiver/ArchiverConfig.json index 40a9776..c27e762 100644 --- a/sbndqm/Archiver/ArchiverConfig.json +++ b/sbndqm/Archiver/ArchiverConfig.json @@ -10,7 +10,8 @@ "redis": { "hostname": "icarus-db02.fnal.gov", "monitor_time": 3, - "port": 6379, "timeout": 30 + "port": 6379, + "timeout": 30 }, "logging": { "directory": "/daq/log/archiver/", diff --git a/sbndqm/Archiver/CreateMonitorMap.py b/sbndqm/Archiver/CreateMonitorMap.py deleted file mode 100644 index 946e11f..0000000 --- a/sbndqm/Archiver/CreateMonitorMap.py +++ /dev/null @@ -1,79 +0,0 @@ -import psycopg2 - -Database = psycopg2.connect(host='icarus-db01.fnal.gov', database='icarus_online_prd', user='runcon_admin', password='xqr98CEm', port=5434) -Cursor = Database.cursor() - - -########################################## -## DROP And CREATE PMT RMS table -########################################## - -#Cursor.execute('DELETE FROM runcon_prd.pmt_channel_rms_median_monitor;') -#Cursor.execute('CREATE TABLE runcon_prd.pmt_channel_rms_median_monitor (CHANNEL_ID INTEGER NOT NULL, SMPL_TIME TIMESTAMP WITH TIME ZONE NOT NULL, SMPL_VALUE DOUBLE PRECISION, PRIMARY KEY (CHANNEL_ID, SMPL_TIME));') -#Database.commit() - -########################################## -## DROP And CREATE PMT baseline table -########################################## - -#Cursor.execute('DELETE FROM runcon_prd.pmt_channel_baseline_mean_monitor;') -#Cursor.execute('CREATE TABLE runcon_prd.pmt_channel_baseline_mean_monitor (CHANNEL_ID INTEGER NOT NULL, SMPL_TIME TIMESTAMP WITH TIME ZONE NOT NULL, SMPL_VALUE DOUBLE PRECISION, PRIMARY KEY (CHANNEL_ID, SMPL_TIME));') -#Database.commit() - - -########################################## -## Clean (DELETE) runcon_prd.monitor_map -########################################## - -#Cursor.execute('DELETE FROM runcon_prd.monitor_map;') -#Database.commit() - -########################################## -## ADD TPC RMS Metrics to monitoring -########################################## - -#for i in range(0,55296): -# Cursor.execute('INSERT INTO runcon_prd.monitor_map (GROUP_NAME, METRIC, CHANNEL_ID, MONITOR_TYPE, POSTGRES_TABLE, TIME_INTERVAL) VALUES (\'tpc_channel\', \'rms\',' + str(i) + ', 0, \'tpc_channel_rms_mean_monitor\', 1200);') -# Database.commit() - -########################################## -## ADD PMT RMS Metrics to monitoring -########################################## - -#for i in range(1,360): -# Cursor.execute('INSERT INTO runcon_prd.monitor_map (GROUP_NAME, METRIC, CHANNEL_ID, MONITOR_TYPE, POSTGRES_TABLE, TIME_INTERVAL) VALUES (\'PMT\', \'rms\',' + str(i) + ', 1, \'pmt_channel_rms_median_monitor\', 3600);') -# Database.commit() - -########################################## -## ADD PMT baseline Metrics to monitoring -########################################## - -#for i in range(0,360): -# Cursor.execute('INSERT INTO runcon_prd.monitor_map (GROUP_NAME, METRIC, CHANNEL_ID, MONITOR_TYPE, POSTGRES_TABLE, TIME_INTERVAL) VALUES (\'PMT\', \'baseline\',' + str(i) + ', 0, \'pmt_channel_baseline_mean_monitor\', 3600);') -# Database.commit() - -########################################## -## Example fetch and print -########################################## - -#Cursor.execute('SELECT * FROM monitor_map;') -#StreamConfig = Cursor.fetchall() -#for i in range(25): -# print(StreamConfig[i]) - -########################################## -## DROP And CREATE TPC RMS table (OLD) -########################################## - -#Cursor.execute('DROP TABLE tpc_channel_rms_mean_monitor') -#Cursor.execute('CREATE TABLE tpc_channel_rms_mean_monitor (CHANNEL_ID INTEGER NOT NULL, SMPL_TIME TIMESTAMP WITH TIME ZONE NOT NULL, SMPL_VALUE DOUBLE PRECISION, PRIMARY KEY (CHANNEL_ID, SMPL_TIME));') - -########################################## -## Create Monitor Map (OLD) -########################################## - -#Cursor.execute('DROP TABLE monitor_map;') -#Database.commit() - -#Cursor.execute('CREATE TABLE monitor_map (GROUP_NAME CHARACTER VARYING(128) NOT NULL, METRIC CHARACTER VARYING(32) NOT NULL, CHANNEL_ID INTEGER NOT NULL, MONITOR_TYPE INTEGER NOT NULL, POSTGRES_TABLE CHARACTER VARYING(128) NOT NULL, TIME_INTERVAL DOUBLE PRECISION NOT NULL, CREATE_TIME TIMESTAMP WITH TIME ZONE DEFAULT NOW(), PRIMARY KEY (GROUP_NAME,METRIC,CHANNEL_ID,MONITOR_TYPE));') -#Database.commit() diff --git a/sbndqm/Archiver/DumpPostgres.py b/sbndqm/Archiver/DumpPostgres.py deleted file mode 100644 index 41b21d3..0000000 --- a/sbndqm/Archiver/DumpPostgres.py +++ /dev/null @@ -1,78 +0,0 @@ -import psycopg2 -import datetime as dt -import numpy as np -import pandas as pd -import time - -def Plane(Channel): - PChannel = Channel % 13824 - p = 0 - if PChannel > 2303 and PChannel < 8064: p = 1 - elif PChannel >= 8064: p = 2 - return p + 3*int(Channel / 13824) - -def HalfCrate(Channel): - return int(Channel / 288) - -def main(): - Granularity = 24 - Database = psycopg2.connect(host='icarus-db01.fnal.gov', database='icarus_online_prd', user='mueller', port=5434) - - Today = dt.datetime.today() - for day in range(0, 90): - Lower = Today - dt.timedelta(days=day) - Upper = Lower + dt.timedelta(days=1) - LS = Lower.strftime('%Y-%m-%d') - US = Upper.strftime('%Y-%m-%d') - Cursor = Database.cursor() - #Cursor.execute("SELECT * FROM runcon_prd.tpc_channel_rms_mean_monitor WHERE smpl_time > \'2021-03-" + "{:0>2}".format(day) + "\' AND smpl_time < \'2021-04-" + "{:0>2}".format(1) + "\';") - Cursor.execute("SELECT * FROM runcon_prd.tpc_channel_rms_mean_monitor WHERE smpl_time > \'" + LS + "\' AND smpl_time < \'" + US + "\' AND channel_id > 4288 AND channel_id < 4319;") - Entries = Cursor.fetchall() - Cursor.close() - print(len(Entries)) - Data = {'channel': [], 'date': [], 'rms': [], 'halfcrate': []} # 'plane': []} - for i in range(len(Entries)): - Data['channel'].append(Entries[i][0]) - Data['date'].append(Entries[i][1]) - Data['rms'].append(Entries[i][2]) - #Data['plane'].append(Plane(Entries[i][0])) - Data['halfcrate'].append(HalfCrate(Entries[i][0])) - dataframe = pd.DataFrame(Data) - dataframe = FilterZeros(dataframe) - - CompressedDataFrame = pd.DataFrame() - for Date in np.unique([dt.datetime(int(x.strftime('%Y')), int(x.strftime('%m')), int(x.strftime('%d'))) for x in dataframe.date.to_numpy()]): - Time = time.mktime(Date.timetuple()) + 60*30*(24.0/Granularity) - DateCut = SelectByDay(Date, dataframe) - for Hour in range(Granularity): - print('Working on hour: ' + str(Hour) + '...') - BinCut = SelectByHour(Hour, DateCut, Difference=(24.0/Granularity)) - #TempDF = BinCut.groupby('channel').mean().reset_index() - #TempDF = BinCut.groupby('plane').mean().reset_index() - TempDF = BinCut.groupby('halfcrate').mean().reset_index() - TempDF['date'] = Time + Hour*(60 * (24.0/Granularity))*60 - CompressedDataFrame = CompressedDataFrame.append(TempDF) - - CompressedDataFrame.to_csv('Run0_' + str(day) + '.csv', index=False) - #dataframe.to_csv('dump.csv') - -def SelectByDay(Date, Data): - print(Date) - AsArray = Data.date.to_numpy() - Cut = Data.loc[[(x.strftime('%d') == Date.strftime('%d') and x.strftime('%m') == Date.strftime('%m') and x.strftime('%Y') == Date.strftime('%Y')) for x in AsArray]] - return Cut - - -def SelectByHour(Hour, Data, Difference=1): - LowHour = Hour*Difference - HighHour = (Hour+1)*Difference - AsArray = Data.date.to_numpy() - Cut = Data.loc[[(int(x.strftime('%-H')) + float(x.strftime('%M'))/60 < HighHour and int(x.strftime('%-H')) + float(x.strftime('%M'))/60 >= LowHour) for x in AsArray]] - return Cut - -def FilterZeros(Data): - Cut = Data.loc[Data.rms != 0] - return Cut - -if __name__ == '__main__': - main() diff --git a/sbndqm/Archiver/inspect_channel.py b/sbndqm/Archiver/inspect_channel.py deleted file mode 100644 index ef2fd10..0000000 --- a/sbndqm/Archiver/inspect_channel.py +++ /dev/null @@ -1,55 +0,0 @@ -import numpy as np -import psycopg2 -import pandas as pd -import datetime as dt -import time - -def query_channel(channel): - database = psycopg2.connect(host='icarus-db01.fnal.gov', - database='icarus_online_prd', - user='mueller', - port=5434) - cursor = database.cursor() - - today = dt.datetime.today() - lower = (today - dt.timedelta(weeks=208)) - upper = today.strftime('%Y-%m-%d') - lower = lower.strftime('%Y-%m-%d') - - table = "runcon_prd.tpc_channel_rms_mean_monitor" - date = "smpl_time > \'" + lower + "\' AND smpl_time < \'" + upper +"\'" - entries = list() - chan = "channel_id = " + str(channel) - comm = "SELECT * FROM " + table + " WHERE " + date + " AND " + chan + ";" - cursor.execute(comm) - entries.extend(cursor.fetchall()) - cursor.close() - entries = list(zip(*entries)) - data = {'channel': entries[0], - 'time': [int(time.mktime(d.timetuple())) for d in entries[1]], - 'rms': entries[2]} - return pd.DataFrame(data) - - -def main(): - df = query_channel(43518) - df.to_csv('ch43518.csv', index=False) - """ - for w in range(4, 104, 4): - print('Querying month: ' + str(int(w/4)) + '.') - df = query_channel(unstable_channels, weeks=w) - print(len(df)) - if w != 4: - pd.concat([pd.read_csv('unstable.csv'), df], - ignore_index=True).to_csv('unstable.csv', index=False) - else: df.to_csv('unstable.csv', index=False) - df = query_channel(stable_channels, weeks=w) - print(len(df)) - if w != 4: - pd.concat([pd.read_csv('stable.csv'), df], - ignore_index=True).to_csv('stable.csv', index=False) - else: df.to_csv('stable.csv', index=False) - """ - -if __name__ == '__main__': - main() diff --git a/sbndqm/Archiver/query_channels.py b/sbndqm/Archiver/query_channels.py deleted file mode 100644 index 5cd9a9a..0000000 --- a/sbndqm/Archiver/query_channels.py +++ /dev/null @@ -1,74 +0,0 @@ -import numpy as np -import psycopg2 -import pandas as pd -import datetime as dt -import time - -def query_channel(channels, weeks=4): - database = psycopg2.connect(host='icarus-db01.fnal.gov', - database='icarus_online_prd', - user='mueller', - port=5434) - cursor = database.cursor() - - today = dt.datetime.today() - lower = (today - dt.timedelta(weeks=weeks)) - upper = (lower + dt.timedelta(weeks=4)).strftime('%Y-%m-%d') - lower = lower.strftime('%Y-%m-%d') - - table = "runcon_prd.tpc_channel_rms_mean_monitor" - date = "smpl_time > \'" + lower + "\' AND smpl_time < \'" + upper +"\'" - entries = list() - for c in channels: - chan = "channel_id = " + str(c) - comm = "SELECT * FROM " + table + " WHERE " + date + " AND " + chan + ";" - cursor.execute(comm) - entries.extend(cursor.fetchall()) - cursor.close() - entries = list(zip(*entries)) - data = {'channel': entries[0], - 'time': [int(time.mktime(d.timetuple())) for d in entries[1]], - 'rms': entries[2]} - return pd.DataFrame(data) - - -def main(): - unstable_channels = [144,432,720,976,1296,1584,3312,3888,4464,4752, - 5040,5328,5904,6192,7056,7920,9936,10224,10512, - 10800,11088,11376,11664,13104,13392,13648,13968, - 14256,16848,17712,18000,18288,18864,19152,20016, - 20592,21168,21456,21712,21744,23152,24048,24624, - 27504,27792,28080,28368,28624,28912,28944,29200, - 329232,31248,31536,32112,32976,33264,34704,34992, - 35280,36720,38448,38736,41040,41616,41904,43056, - 44208,44496,44784,45360,45616,46224,46512,46800, - 47664,47952,48816,49104,49360,49712,50256,51696, - 54000,55152] - stable_channels = [16,304,592,848,1168,1456,3184,3760,4336,4624,4912, - 5200,5776,6064,6928,7792,9808,10096,10384,10672, - 10960,11248,11536,12976,13264,13520,13840,14128, - 16720,17584,17872,18160,18736,19024,19888,20464, - 21040,21328,21584,21616,23024,23920,24496,27376, - 27664,27952,28240,28496,28784,28816,29072,29104, - 31120,31408,31984,32848,33136,34576,34864,35152, - 36592,38320,38608,40912,41488,41776,42928,44080, - 44368,44656,45232,45488,46096,46384,46672,47536, - 47824,48688,48976,49232,49584,50128,51568,53872, - 55024] - for w in range(4, 104, 4): - print('Querying month: ' + str(int(w/4)) + '.') - df = query_channel(unstable_channels, weeks=w) - print(len(df)) - if w != 4: - pd.concat([pd.read_csv('unstable.csv'), df], - ignore_index=True).to_csv('unstable.csv', index=False) - else: df.to_csv('unstable.csv', index=False) - df = query_channel(stable_channels, weeks=w) - print(len(df)) - if w != 4: - pd.concat([pd.read_csv('stable.csv'), df], - ignore_index=True).to_csv('stable.csv', index=False) - else: df.to_csv('stable.csv', index=False) - -if __name__ == '__main__': - main() From aaca5a8fe005aa50e32ba07e5952a478be09c704 Mon Sep 17 00:00:00 2001 From: Matteo Vicenzi Date: Wed, 17 Sep 2025 17:19:13 -0400 Subject: [PATCH 3/9] remove attempted rewrite --- sbndqm/Archiver/NewArchiver.py | 398 --------------------------------- 1 file changed, 398 deletions(-) delete mode 100644 sbndqm/Archiver/NewArchiver.py diff --git a/sbndqm/Archiver/NewArchiver.py b/sbndqm/Archiver/NewArchiver.py deleted file mode 100644 index a7535be..0000000 --- a/sbndqm/Archiver/NewArchiver.py +++ /dev/null @@ -1,398 +0,0 @@ -import argparse -import redis as rd -import json -import time -import signal -import sys -import psutil -import struct -import psycopg2 -import logging -from multiprocessing import Process -import math - -def connect_redis(cfg, args): - redis = rd.Redis(host=cfg['redis']['hostname'], - port=cfg['redis']['port'], - password=args.password) - try: - redis.ping() - except: - logging.info('ERROR: trouble connecting to the Redis database on server (%s) port (%i)' % (cfg['redis']['hostname'])) - -def ConnectRedis(Config, args): - r = redis.Redis(host=Config['redis']['hostname'], port=Config['redis']['port'], password=args.password) - try: - r.ping() - except: - print('Archiver Error: trouble connecting to Redis database on server (%s) port (%i)' % (Config['redis']['hostname'], Config['redis']['port'])) - logging.info('ERROR: trouble connecting to the Redis database on server (%s) port (%i)' % (Config['redis']['hostname'], Config['redis']['port'])) - return None - logging.info('Connection to the Redis database has been successfully established on server (%s) port (%i)' % (Config['redis']['hostname'], Config['redis']['port'])) - return r - -def ConnectPostgreSQL(Config): - with open(Config['postgresql']['passfile'], 'r') as passfile: - pwd = passfile.read().replace('\n', '') - p = psycopg2.connect(host=Config['postgresql']['hostname'], database=Config['postgresql']['database'], user=Config['postgresql']['user'], port=Config['postgresql']['port'], password=pwd) - - try: - cur = p.cursor() - cur.execute('SELECT 1') - except psycopg2.OperationalError: - print('Archiver Error: trouble connecting to PostgreSQL database on server (%s) port (%i)' % (Config['postgresql']['hostname'], Config['postgresql']['port'])) - logging.info('ERROR: trouble connecting to PostgreSQL database on server (%s) port (%i)' % (Config['postgresql']['hostname'], Config['postgresql']['port'])) - return None, None - logging.info('Connection to the PostgreSQL database has been successfully established on server (%s) port (%i)' % (Config['postgresql']['hostname'], Config['postgresql']['port'])) - - return p, cur - -def LoadConfig(Name): - with open(Name) as JSONFile: - Config = json.load(JSONFile) - - mode = 'a' - if(Config['logging']['overwrite'] == True): mode = 'w' - logging.basicConfig(filename=Config['logging']['directory'] + Config['logging']['name'], - level=logging.INFO, - filemode=mode, - format='%(asctime)s - %(message)s', - datefmt='%Y-%m-%d %H:%M:%S') - return Config - -def main(args): - - signal.signal(signal.SIGINT, signal_handler) - - #Load configuration information from the configuration file. - Config = LoadConfig('ArchiverConfig.json') - logging.info('Configuration loaded. Starting Archiver...') - - #Connect to the Redis database. - r = ConnectRedis(Config, args) - if r is None: - sys.exit(0) - - #Connect to the PostgreSQL database. - p, cur = ConnectPostgreSQL(Config) - if cur is None: - sys.exit(0) - - - #This next bit deletes the record of streams that have been completed. This is for testing purposes only. - #for key in r.scan_iter('PMT*_LatestCompleted'): - # r.delete(key) - - #for key in r.scan_iter("*_LatestCompleted"): - #print(key) - #if key == 'tpc_channel:1259:rms:archiving': - # print(key[:-16], 'Reseting') - #CurrentTime = int(GetLatest(r,key[:-16]).split('-')[0]) - #print('Current time', CurrentTime) - #SetLatest(r,key[:-16], '1596208000000') - #SetLatest(r,key[:-16], '1608237252732') - #SetLatest(r,key[:-16], '1648073157000') - - #Now we split the job into multiple processes that each handle some portion of the streams. - try: - cur.execute('SELECT * FROM ' + Config['postgresql']['metric_config'] + ';') - StreamConfig = cur.fetchall() - except Exception as err: - logging.error('Error while querying metric config table: {}'.format(err)) - logging.error(type(err)) - sys.exit(0) - ProcessCount = int(args.processes) - ProcessList = [] - #Start the Monitor process, which monitors CPU usage, memory, and sends a heartbeat to Redis. - Process(target=Monitor, args=(r, Config)).start() - for i in range(ProcessCount): - KeyList = [ x[0] + ':' + str(x[2]) + ':' + x[1] + ':' + 'archiving' for x in StreamConfig ][i::ProcessCount] - ProcessList.append(Process(target=ProcessStreams, args=(r, p, cur, { x : GetLatest(r, x) for x in KeyList }, { x[0] + ':' + str(x[2]) + ':' + x[1] + ':' + 'archiving' : x for x in StreamConfig[i::ProcessCount]}, args))) - for i in range(ProcessCount): - ProcessList[i].start() - for i in range(ProcessCount): - ProcessList[i].join() - -def ProcessStreams(r, p, cur, StreamDict, Config, args): - - MetricDict = {x : [] for x in StreamDict.keys()} - - logging.info('Archiver is ready to process Redis streams..') - logging.info('Latest completed for ' + StreamDict.keys()[0] + ': ' + StreamDict[StreamDict.keys()[0]]) - - while True: - TotalSetTime = 0 - TotalSetN = 0 - try: - st = time.time() - ReadStream = r.xread(StreamDict, block=0) - logging.info('XREAD time: {}'.format(time.time()-st)) - except redis.RedisError as err: - logging.error('Error while reading streams: {}'.format(err)) - logging.error(type(err)) - if ReconnectCount == 4: - sys.exit(0) - else: - ReconnectCount += 1 - logging.error('Reconnecting to Redis (Attempt ' + str(ReconnectCount) + ') ...') - ArcConfig = LoadConfig('ArchiverConfig.json') - r = ConnectRedis(Config, args) - continue - - if len(ReadStream) == 0: - logging.error('Block timeout. Reconnecting to Redis...') - ArcConfig = LoadConfig('ArchiverConfig.json') - r = ConnectRedis(Config, args) - continue - - ReconnectCount = 0 - - #Loop over the streams which have entries to be archived. - for StreamObject in ReadStream: - #Loop over the individual entries in the stream. - for DataObject in StreamObject[1]: - #Check if the latest completed archived entry is '0' (no metrics have been archived yet for the stream). - if StreamDict[StreamObject[0]] == '0': - NewLatest = str( int( DataObject[0].split('-')[0] ) - 1 ) + '-0' - TotalSetTime += SetLatest(r, StreamObject[0], NewLatest) - TotalSetN += 1 - ProcessData(MetricDict, DataObject, Config, StreamObject[0]) - StreamDict[StreamObject[0]] = DataObject[0] - #Check if the current entry is within the time block (contained in the interval of entries to be archived). - elif int( DataObject[0].split('-')[0] ) < int( GetLatest(r, StreamObject[0]).split('-')[0] ) + 1000*Config[StreamObject[0]][5]: - ProcessData(MetricDict, DataObject, Config, StreamObject[0]) - StreamDict[StreamObject[0]] = DataObject[0] - #Make sure that there are metrics to archive (No metrics in MetricDict, but GetLatest is more than one time block previous to the next entry). - elif int( DataObject[0].split('-')[0] ) > int( GetLatest(r, StreamObject[0]).split('-')[0] ) + 1000*Config[StreamObject[0]][5] and len(MetricDict[StreamObject[0]]) == 0: - NewLatest = str( int( DataObject[0].split('-')[0] ) - 1 ) + '-0' - TotalSetTime += SetLatest(r, StreamObject[0], NewLatest) - TotalSetN += 1 - ProcessData(MetricDict, DataObject, Config, StreamObject[0]) - StreamDict[StreamObject[0]] = DataObject[0] - #The entry is outside the current time block. Set a new latest completed and write to the database. - else: - #Case for "mean" averaging. - if Config[StreamObject[0]][3] == 0: - TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) - TotalSetN += 1 - WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][1], int(MetricDict[StreamObject[0]][0].split('-')[0])) - MetricDict[StreamObject[0]] = [] - #Case for "median" averaging. - elif Config[StreamObject[0]][3] == 1: - TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][-1][0]) - TotalSetN += 1 - MetricList = [ x[1] for x in MetricDict[StreamObject[0]] ] - WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], median(MetricList), int(MetricDict[StreamObject[0]][-1][0].split('-')[0])) - MetricDict[StreamObject[0]] = [] - #Case for "mode" averaging. - elif Config[StreamObject[0]][3] == 2: - TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) - TotalSetN += 1 - WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][3], int(MetricDict[StreamObject[0]][0].split('-')[0])) - MetricDict[StreamObject[0]] = [] - #Case for "max" averaging. - elif Config[StreamObject[0]][3] == 3: - TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) - TotalSetN += 1 - WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][1], int(MetricDict[StreamObject[0]][0].split('-')[0])) - MetricDict[StreamObject[0]] = [] - #Case for "min" averaging. - elif Config[StreamObject[0]][3] == 4: - TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) - TotalSetN += 1 - WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][1], int(MetricDict[StreamObject[0]][0].split('-')[0])) - MetricDict[StreamObject[0]] = [] - #Case for "last" averaging. - elif Config[StreamObject[0]][3] == 5: - TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) - WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][1], int(MetricDict[StreamObject[0]][0].split('-')[0])) - MetricDict[StreamObject[0]] = [] - #After performing the archiving on the previous block of data, we can now process the current entry. - ProcessData(MetricDict, DataObject, Config, StreamObject[0]) - StreamDict[StreamObject[0]] = DataObject[0] - #Check to see if there is a gap in the data stream (the current object is more than one time interval from the latest completed). - if int( DataObject[0].split('-')[0] ) > int( GetLatest(r, StreamObject[0]).split('-')[0] ) + 1000*Config[StreamObject[0]][5]: - NewLatest = str( int( DataObject[0].split('-')[0] ) - 1 ) + '-0' - TotalSetTime += SetLatest(r, StreamObject[0], NewLatest) - TotalSetN += 1 - logging.info('Total set time: {}'.format(TotalSetTime)) - logging.info('Average set time: {}'.format(float(TotalSetTime)/float(TotalSetN))) - -def ProcessData(Metrics, Datum, Config, StreamName): - Value = float(ReadDatum(Datum[1])) - if not math.isnan(Value): - if Config[StreamName][3] == 0: - if len(Metrics[StreamName]) == 0: - Metrics[StreamName] = [Datum[0], Value, 1] - else: - Metrics[StreamName][0] = Datum[0] - Metrics[StreamName][1] = (Metrics[StreamName][1] * Metrics[StreamName][2] + Value) / (Metrics[StreamName][2] + 1) - Metrics[StreamName][2] += 1 - elif Config[StreamName][3] == 1: #Keeps all data for now. - Metrics[StreamName].append( (Datum[0], Value) ) - elif Config[StreamName][3] == 2: - if len(Metrics[StreamName]) == 0: - Metrics[StreamName] = [Datum[0], {Value : 1}, 1, Value] - else: - Metrics[StreamName][0] = Datum[0] - if Value not in Metrics[StreamName][1]: - Metrics[StreamName][1][Value] = 1 - else: - Metrics[StreamName][1][Value] += 1 - if Metrics[StreamName][1][Value] > Metrics[StreamName][2]: - Metrics[StreamName][2] = Metrics[StreamName][1][Value] - Metrics[StreamName][3] = Value - elif Config[StreamName][3] == 3: - if len(Metrics[StreamName]) == 0: - Metrics[StreamName] = [Datum[0], Value] - else: - Metrics[StreamName][0] = Datum[0] - if Metrics[StreamName][1] < Value: Metrics[StreamName][1] = Value - elif Config[StreamName][3] == 4: - if len(Metrics[StreamName]) == 0: - Metrics[StreamName] = [Datum[0], Value] - else: - Metrics[StreamName][0] = Datum[0] - if Metrics[StreamName][1] > Value: Metrics[StreamName][1] = Value - elif Config[StreamName][3] == 5: - if len(Metrics[StreamName]) == 0: - Metrics[StreamName] = [Datum[0], Value] - else: - Metrics[StreamName][0] = Datum[0] - Metrics[StreamName][1] = Value - -def TypeToStructType(Name): - if Name == "int8_t": return "b" - if Name == "int16_t": return "h" - if Name == "int32_t": return "i" - if Name == "int64_t": return "q" - if Name == "uint8_t": return "B" - if Name == "uint16_t": return "H" - if Name == "uint32_t": return "I" - if Name == "uint64_t": return "Q" - if Name == "float": return "f" - if Name == "double": return "d" - -def TypeToSize(Name): - if Name == "int8_t": return 1 - if Name == "int16_t": return 2 - if Name == "int32_t": return 4 - if Name == "int64_t": return 8 - if Name == "uint8_t": return 1 - if Name == "uint16_t": return 2 - if Name == "uint32_t": return 4 - if Name == "uint64_t": return 8 - if Name == "float": return 4 - if Name == "double": return 8 - -def ParseBinary(Binary, TypeName): - Size = TypeToSize(TypeName) - Form = TypeToStructType(TypeName) - ret = [] - for i in range(len(Binary) / Size): - dat = Binary[i*Size : (i+1)*Size] - ret.append(struct.unpack(Form, dat)[0]) - return ret - -def ReadDatum(dat): - for key, val in dat.items(): - if key == 'val' or key == 'dat': return val - return ParseBinary(val,key)[0] - -def GetLatest(Database, StreamName): - try: - Latest = Database.get(StreamName + '_LatestCompleted') - except redis.RedisError as err: - logging.error(str('Error while getting latest for stream ' + StreamName + ': {}').format(err)) - logging.error(type(err)) - sys.exit(0) - if Latest is not None: return Latest - else: - Database.set(StreamName + '_LatestCompleted','0') - return '0' - -def SetLatest(Database, StreamName, Latest): - s = time.time() - try: - Database.set(StreamName + '_LatestCompleted', Latest) - except redis.RedisError as err: - logging.error(str('Error while setting latest for stream ' + StreamName + ' and timestamp ' + Latest + ': {}').format(err)) - logging.error(type(err)) - sys.exit(0) - return time.time() - s - -def median(l): - l = sorted(l) - Length = len(l) - if Length < 1: - return None - if Length % 2 == 0: - return ( l[(Length-1)/2] + l[(Length+1)/2] ) / 2.0 - else: - return l[(Length-1)/2] - -def WritePostgres(p, cur, Table, Channel, Value, Time): - Timestamp = int(Time/1000) - Attempt = 0 - if math.isnan(Value): - Value = 0; - Command = 'INSERT INTO runcon_prd.' + Table + ' (CHANNEL_ID, SMPL_TIME, SMPL_VALUE) VALUES (' + str(Channel) + ',to_timestamp(' + str(Timestamp) + '),' + str(Value) + ');' - while Attempt < 5: - try: - Attempt += 1 - cur.execute(Command) - p.commit() - except psycopg2.OperationalError as err: - time.sleep(60) - logging.error(str('Attempt ' + str(Attempt) + ': Possible PostgreSQL connection issue. Re-connecting...')) - Config = LoadConfig('ArchiverConfig.json') - p, cur = ConnectPostgreSQL(Config) - if cur is None: - sys.exit(0) - except Exception as err: - logging.error(str('Attempt ' + str(Attempt) + ': Error while archiving to table ' + Table + ' for channel ' + str(Channel) + ' and value ' + str(Value) + 'with timestamp ' + str(Time) + ': {}').format(err)) - logging.error(type(err)) - time.sleep(60) - #sys.exit(0) - else: break - else: - logging.error('Failed reattempts; exiting...') - sys.exit(0) - -def Monitor(r, Config): - logging.info('Monitor is running.') - ErrorCount = 0 - while ErrorCount < 4: - pipeline = r.pipeline() - #pipeline.xadd('archiver_heartbeat', {'float': struct.pack('f', 0)}, maxlen=1) - #CPUs = psutil.cpu_percent(interval=Config['redis']['monitor_time'], percpu=True) - #CPUAverage = psutil.cpu_percent(interval=Config['redis']['monitor_time']) - #Mem = psutil.virtual_memory()[2] - #try: - # for i in range(len(CPUs)): pipeline.xadd('archiver_cpu'+str(i), {'float': struct.pack('f', CPUs[i])}, maxlen=10) - # pipeline.xadd('archiver_cpu_average', {'float': struct.pack('f', CPUAverage)}, maxlen=10) - # pipeline.xadd('archiver_mem', {'float': struct.pack('f', Mem)}, maxlen=10) - # [ _ for _ in pipeline.execute()] - #except redis.RedisError as err: - # logging.error(str('Error while executing pipeline for archiver metrics: {}').format(err)) - # sys.exit(0) - #time.sleep(300) - try: - pipeline.xadd('archiver_heartbeat', {'float': struct.pack('f', 0)}, maxlen=1) - [ _ for _ in pipeline.execute() ] - except redis.RedisError as err: - logging.error(str('Error while executing pipeline for archiver heartbeat: {}').format(err)) - #sys.exit(0) - ErrorCount += 1 - time.sleep(60) - -def signal_handler(sig, frame): - print('Keyboard Interrupt') - logging.info('Keyboard interrupt. Exiting Archiver.') - sys.exit(0) - -if __name__ == "__main__": - args = argparse.ArgumentParser() - args.add_argument("-pw", "--password", default=None) - args.add_argument("-pr", "--processes", default=1) - main(args.parse_args()) From debff49f282636bdc2ac44bbca65176a39cd6126 Mon Sep 17 00:00:00 2001 From: Matteo Vicenzi Date: Mon, 22 Sep 2025 16:22:21 -0400 Subject: [PATCH 4/9] python3 conversion by Kiyoung --- sbndqm/Archiver/Archiver.py | 148 ++++++++++++++++++++---------------- 1 file changed, 83 insertions(+), 65 deletions(-) diff --git a/sbndqm/Archiver/Archiver.py b/sbndqm/Archiver/Archiver.py index 4faf073..b6d194f 100644 --- a/sbndqm/Archiver/Archiver.py +++ b/sbndqm/Archiver/Archiver.py @@ -15,7 +15,7 @@ def ConnectRedis(Config, args): r = redis.Redis(host=Config['redis']['hostname'], port=Config['redis']['port'], password=args.password) try: r.ping() - except: + except Exception: print('Archiver Error: trouble connecting to Redis database on server (%s) port (%i)' % (Config['redis']['hostname'], Config['redis']['port'])) logging.info('ERROR: trouble connecting to the Redis database on server (%s) port (%i)' % (Config['redis']['hostname'], Config['redis']['port'])) return None @@ -44,7 +44,8 @@ def LoadConfig(Name): Config = json.load(JSONFile) mode = 'a' - if(Config['logging']['overwrite'] == True): mode = 'w' + if Config['logging'].get('overwrite') == True: + mode = 'w' logging.basicConfig(filename=Config['logging']['directory'] + Config['logging']['name'], level=logging.INFO, filemode=mode, @@ -70,7 +71,6 @@ def main(args): if cur is None: sys.exit(0) - #This next bit deletes the record of streams that have been completed. This is for testing purposes only. #for key in r.scan_iter('PMT*_LatestCompleted'): # r.delete(key) @@ -98,7 +98,10 @@ def main(args): Process(target=Monitor, args=(r, Config)).start() for i in range(ProcessCount): KeyList = [ x[0] + ':' + str(x[2]) + ':' + x[1] + ':' + 'archiving' for x in StreamConfig ][i::ProcessCount] - ProcessList.append(Process(target=ProcessStreams, args=(r, p, cur, { x : GetLatest(r, x) for x in KeyList }, { x[0] + ':' + str(x[2]) + ':' + x[1] + ':' + 'archiving' : x for x in StreamConfig[i::ProcessCount]}, args))) + stream_last = { x : GetLatest(r, x) for x in KeyList } + per_stream_cfg = { x[0] + ':' + str(x[2]) + ':' + x[1] + ':' + 'archiving' : x for x in StreamConfig[i::ProcessCount]} + ProcessList.append(Process(target=ProcessStreams, args=(r, p, cur, stream_last, per_stream_cfg, args))) + for i in range(ProcessCount): ProcessList[i].start() for i in range(ProcessCount): @@ -109,11 +112,18 @@ def ProcessStreams(r, p, cur, StreamDict, Config, args): MetricDict = { x : [] for x in StreamDict.keys()} logging.info('Archiver in ProcessStreams.') - logging.info('LatestCompleted for ' + StreamDict.keys()[0] + ': ' + StreamDict[StreamDict.keys()[0]]) + first_key = next(iter(StreamDict)) if StreamDict else None + if first_key is not None: + logging.info('LatestCompleted for ' + first_key + ': ' + StreamDict[first_key]) + else: + logging.info('No streams assigned to this process.') + return + + ReconnectCount = 0 #We now begin to continuously query these streams for new metrics, archiving them when possible. while True: - TotalSetTime = 0 + TotalSetTime = 0.0 TotalSetN = 0 #ReadStream is a list of stream object which have new entries that have yet to be archived. try: @@ -130,87 +140,93 @@ def ProcessStreams(r, p, cur, StreamDict, Config, args): ReconnectCount += 1 logging.error('Reconnecting to Redis (Attempt ' + str(ReconnectCount) + ') ...') ArcConfig = LoadConfig('ArchiverConfig.json') - r = ConnectRedis(Config, args) + r = ConnectRedis(ArcConfig, args) continue if len(ReadStream) == 0: logging.error('Block timeout. Reconnecting to Redis...') ArcConfig = LoadConfig('ArchiverConfig.json') - r = ConnectRedis(Config, args) + r = ConnectRedis(ArcConfig, args) continue ReconnectCount = 0 #Loop over the streams which have entries to be archived. for StreamObject in ReadStream: + stream_name_b = StreamObject[0] + stream_name = stream_name_b.decode('utf-8') + entries = StreamObject[1] #Loop over the individual entries in the stream. - for DataObject in StreamObject[1]: + for DataObject in entries: + entry_id_b, entry_fields = DataObject + entry_id = entry_id_b.decode('utf-8') + #Check if the latest completed archived entry is '0' (no metrics have been archived yet for the stream). - if StreamDict[StreamObject[0]] == '0': - NewLatest = str( int( DataObject[0].split('-')[0] ) - 1 ) + '-0' - TotalSetTime += SetLatest(r, StreamObject[0], NewLatest) + if StreamDict[stream_name] == '0': + NewLatest = str( int( entry_id.split('-')[0] ) - 1 ) + '-0' + TotalSetTime += SetLatest(r, stream_name, NewLatest) TotalSetN += 1 - ProcessData(MetricDict, DataObject, Config, StreamObject[0]) - StreamDict[StreamObject[0]] = DataObject[0] + ProcessData(MetricDict, (entry_id, entry_fields), Config, stream_name) + StreamDict[stream_name] = entry_id #Check if the current entry is within the time block (contained in the interval of entries to be archived). - elif int( DataObject[0].split('-')[0] ) < int( GetLatest(r, StreamObject[0]).split('-')[0] ) + 1000*Config[StreamObject[0]][5]: - ProcessData(MetricDict, DataObject, Config, StreamObject[0]) - StreamDict[StreamObject[0]] = DataObject[0] + elif int( entry_id.split('-')[0] ) < int( GetLatest(r, stream_name).split('-')[0] ) + 1000*Config[stream_name][5]: + ProcessData(MetricDict, (entry_id, entry_fields), Config, stream_name) + StreamDict[stream_name] = entry_id #Make sure that there are metrics to archive (No metrics in MetricDict, but GetLatest is more than one time block previous to the next entry). - elif int( DataObject[0].split('-')[0] ) > int( GetLatest(r, StreamObject[0]).split('-')[0] ) + 1000*Config[StreamObject[0]][5] and len(MetricDict[StreamObject[0]]) == 0: - NewLatest = str( int( DataObject[0].split('-')[0] ) - 1 ) + '-0' - TotalSetTime += SetLatest(r, StreamObject[0], NewLatest) + elif int( entry_id.split('-')[0] ) > int( GetLatest(r, stream_name).split('-')[0] ) + 1000*Config[stream_name][5] and len(MetricDict[stream_name]) == 0: + NewLatest = str( int( entry_id.split('-')[0] ) - 1 ) + '-0' + TotalSetTime += SetLatest(r, stream_name, NewLatest) TotalSetN += 1 - ProcessData(MetricDict, DataObject, Config, StreamObject[0]) - StreamDict[StreamObject[0]] = DataObject[0] + ProcessData(MetricDict, (entry_id, entry_fields), Config, stream_name) + StreamDict[stream_name] = entry_id #The entry is outside the current time block. Set a new latest completed and write to the database. - else: + else: #Case for "mean" averaging. - if Config[StreamObject[0]][3] == 0: - TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + if Config[stream_name][3] == 0: + TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][0]) TotalSetN += 1 - WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][1], int(MetricDict[StreamObject[0]][0].split('-')[0])) - MetricDict[StreamObject[0]] = [] + WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], MetricDict[stream_name][1], int(MetricDict[stream_name][0].split('-')[0])) + MetricDict[stream_name] = [] #Case for "median" averaging. - elif Config[StreamObject[0]][3] == 1: - TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][-1][0]) + elif Config[stream_name][3] == 1: + TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][-1][0]) TotalSetN += 1 - MetricList = [ x[1] for x in MetricDict[StreamObject[0]] ] - WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], median(MetricList), int(MetricDict[StreamObject[0]][-1][0].split('-')[0])) - MetricDict[StreamObject[0]] = [] + MetricList = [ x[1] for x in MetricDict[stream_name] ] + WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], median(MetricList), int(MetricDict[stream_name][-1][0].split('-')[0])) + MetricDict[stream_name] = [] #Case for "mode" averaging. - elif Config[StreamObject[0]][3] == 2: - TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + elif Config[stream_name][3] == 2: + TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][0]) TotalSetN += 1 - WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][3], int(MetricDict[StreamObject[0]][0].split('-')[0])) - MetricDict[StreamObject[0]] = [] + WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], MetricDict[stream_name][3], int(MetricDict[stream_name][0].split('-')[0])) + MetricDict[stream_name] = [] #Case for "max" averaging. - elif Config[StreamObject[0]][3] == 3: - TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + elif Config[stream_name][3] == 3: + TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][0]) TotalSetN += 1 - WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][1], int(MetricDict[StreamObject[0]][0].split('-')[0])) - MetricDict[StreamObject[0]] = [] + WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], MetricDict[stream_name][1], int(MetricDict[stream_name][0].split('-')[0])) + MetricDict[stream_name] = [] #Case for "min" averaging. - elif Config[StreamObject[0]][3] == 4: - TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) + elif Config[stream_name][3] == 4: + TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][0]) TotalSetN += 1 - WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][1], int(MetricDict[StreamObject[0]][0].split('-')[0])) - MetricDict[StreamObject[0]] = [] + WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], MetricDict[stream_name][1], int(MetricDict[stream_name][0].split('-')[0])) + MetricDict[stream_name] = [] #Case for "last" averaging. - elif Config[StreamObject[0]][3] == 5: - TotalSetTime += SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0]) - WritePostgres(p, cur, Config[StreamObject[0]][4], Config[StreamObject[0]][2], MetricDict[StreamObject[0]][1], int(MetricDict[StreamObject[0]][0].split('-')[0])) - MetricDict[StreamObject[0]] = [] + elif Config[stream_name][3] == 5: + TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][0]) + WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], MetricDict[stream_name][1], int(MetricDict[stream_name][0].split('-')[0])) + MetricDict[stream_name] = [] #After performing the archiving on the previous block of data, we can now process the current entry. - ProcessData(MetricDict, DataObject, Config, StreamObject[0]) - StreamDict[StreamObject[0]] = DataObject[0] + ProcessData(MetricDict, (entry_id, entry_fields), Config, stream_name) + StreamDict[stream_name] = entry_id #Check to see if there is a gap in the data stream (the current object is more than one time interval from the latest completed). - if int( DataObject[0].split('-')[0] ) > int( GetLatest(r, StreamObject[0]).split('-')[0] ) + 1000*Config[StreamObject[0]][5]: - NewLatest = str( int( DataObject[0].split('-')[0] ) - 1 ) + '-0' - TotalSetTime += SetLatest(r, StreamObject[0], NewLatest) + if int( entry_id.split('-')[0] ) > int( GetLatest(r, stream_name).split('-')[0] ) + 1000*Config[stream_name][5]: + NewLatest = str( int( entry_id.split('-')[0] ) - 1 ) + '-0' + TotalSetTime += SetLatest(r, stream_name, NewLatest) TotalSetN += 1 logging.info('Total set time: {}'.format(TotalSetTime)) - logging.info('Average set time: {}'.format(float(TotalSetTime)/float(TotalSetN) if TotalSetN else 0)) + logging.info('Average set time: {}'.format(float(TotalSetTime)/float(TotalSetN) if TotalSetN else 0.0)) def ProcessData(Metrics, Datum, Config, StreamName): Value = float(ReadDatum(Datum[1])) @@ -283,15 +299,17 @@ def ParseBinary(Binary, TypeName): Size = TypeToSize(TypeName) Form = TypeToStructType(TypeName) ret = [] - for i in range(len(Binary) / Size): + for i in range(len(Binary) // Size): dat = Binary[i*Size : (i+1)*Size] ret.append(struct.unpack(Form, dat)[0]) return ret def ReadDatum(dat): for key, val in dat.items(): - if key == 'val' or key == 'dat': return val - return ParseBinary(val,key)[0] + if key == b'val' or key == b'dat': + return val + type_name = key.decode('utf-8') + return ParseBinary(val, type_name)[0] def GetLatest(Database, StreamName): try: @@ -300,7 +318,7 @@ def GetLatest(Database, StreamName): logging.error(str('Error while getting latest for stream ' + StreamName + ': {}').format(err)) logging.error(type(err)) sys.exit(0) - if Latest is not None: return Latest + if Latest is not None: return Latest.decode('utf-8') else: Database.set(StreamName + '_LatestCompleted','0') return '0' @@ -321,9 +339,9 @@ def median(l): if Length < 1: return None if Length % 2 == 0: - return ( l[(Length-1)/2] + l[(Length+1)/2] ) / 2.0 + return ( l[(Length-1)//2] + l[(Length+1)//2] ) / 2.0 else: - return l[(Length-1)/2] + return l[(Length-1)//2] def WritePostgres(p, cur, Table, Channel, Value, Time): Timestamp = int(Time/1000) @@ -372,7 +390,7 @@ def Monitor(r, Config): # sys.exit(0) #time.sleep(300) try: - pipeline.xadd('archiver_heartbeat', {'float': struct.pack('f', 0)}, maxlen=1) + pipeline.xadd('archiver_heartbeat', {'float': struct.pack('f', 0.0)}, maxlen=1) [ _ for _ in pipeline.execute() ] except redis.RedisError as err: logging.error(str('Error while executing pipeline for archiver heartbeat: {}').format(err)) @@ -386,7 +404,7 @@ def signal_handler(sig, frame): sys.exit(0) if __name__ == "__main__": - args = argparse.ArgumentParser() - args.add_argument("-pw", "--password", default=None) - args.add_argument("-pr", "--processes", default=1) - main(args.parse_args()) + parser = argparse.ArgumentParser() + parser.add_argument("-pw", "--password", default=None) + parser.add_argument("-pr", "--processes", default=1) + main(parser.parse_args()) From 8c63c6323dc1b24680bc45ce59df31aa8d019ce8 Mon Sep 17 00:00:00 2001 From: Matteo Vicenzi Date: Mon, 22 Sep 2025 16:37:08 -0400 Subject: [PATCH 5/9] restore newer ddl files" --- sbndqm/Archiver/ddl/channelMonitor.sql | 24 ++++++++--------- sbndqm/Archiver/ddl/createMonitor.sh | 7 +++-- sbndqm/Archiver/ddl/monitorMap.sql | 37 +++++++++++++++----------- 3 files changed, 38 insertions(+), 30 deletions(-) diff --git a/sbndqm/Archiver/ddl/channelMonitor.sql b/sbndqm/Archiver/ddl/channelMonitor.sql index 9058226..24f0e27 100644 --- a/sbndqm/Archiver/ddl/channelMonitor.sql +++ b/sbndqm/Archiver/ddl/channelMonitor.sql @@ -5,11 +5,11 @@ \echo :CH1 \echo :CHN -CREATE OR REPLACE FUNCTION MONMON(GROUP_NAME CHARACTER, - METRIC CHARACTER, - MTYPE INTEGER, - CH1 INTEGER, - CHN INTEGER) +CREATE OR REPLACE FUNCTION RUNCON_PRD.MONMON(GROUP_NAME CHARACTER, + METRIC CHARACTER, + MONITOR_TYPE_NAME CHARACTER, + CH1 INTEGER, + CHN INTEGER) RETURNS CHARACTER VARYING(32) AS $BODY$ DECLARE @@ -18,22 +18,22 @@ $BODY$ MTYPE INTEGER; CTR INTEGER; BEGIN - MONITOR_NAMED := (SELECT TYPE_NAME FROM MONITOR_TYPES WHERE MONITOR_TYPE=MTYPE); - TABLE_NAME := LOWER(GROUP_NAME || '_' || METRIC || '_' || MONITOR_NAMED || '_MONITOR'); + MTYPE := (SELECT MONITOR_TYPE FROM RUNCON_PRD.MONITOR_TYPES WHERE TYPE_NAME=UPPER(MONITOR_TYPE_NAME)); + TABLE_NAME := LOWER(GROUP_NAME || '_' || METRIC || '_' || MONITOR_TYPE_NAME || '_MONITOR'); RAISE NOTICE 'Table Name: %',TABLE_NAME; EXECUTE FORMAT(' - CREATE TABLE %I + CREATE TABLE RUNCON_PRD.%I ( CHANNEL_ID INTEGER NOT NULL, SMPL_TIME TIMESTAMP WITH TIME ZONE NOT NULL, SMPL_VALUE DOUBLE PRECISION, - PRIMARY KEY (CHANNEL_ID,SMPL_VALUE) + PRIMARY KEY (CHANNEL_ID,SMPL_TIME) )',TABLE_NAME); CTR := CH1; WHILE CTR <= CHN LOOP EXECUTE FORMAT(' - INSERT INTO MONITOR_MAP VALUES (%L,%L,%L,%L,%L,3600,NOW()) - ',GROUP_NAME,METRIC,CTR,MTYPE,TABLE_NAME); + INSERT INTO RUNCON_PRD.MONITOR_MAP (GROUP_NAME,METRIC,MONITOR_TYPE,POSTGRES_TABLE,CHANNEL_ID) VALUES (%L,%L,%L,%L,%L) + ',GROUP_NAME,METRIC,MTYPE,TABLE_NAME,CTR); CTR := CTR + 1; END LOOP; RETURN TABLE_NAME; @@ -41,5 +41,5 @@ $BODY$ $BODY$ LANGUAGE plpgsql; -SELECT MONMON(:GROUP_NAME,:METRIC,:MONITOR_TYPE,:CH1,:CHN); +SELECT RUNCON_PRD.MONMON(:GROUP_NAME,:METRIC,:MONITOR_TYPE,:CH1,:CHN); diff --git a/sbndqm/Archiver/ddl/createMonitor.sh b/sbndqm/Archiver/ddl/createMonitor.sh index 7ee35fe..1fc5eee 100755 --- a/sbndqm/Archiver/ddl/createMonitor.sh +++ b/sbndqm/Archiver/ddl/createMonitor.sh @@ -12,7 +12,10 @@ if [ $# -ne 5 ]; then exit fi -psql -U badgett -h cdpgsdev -p 5488 -d sbnteststand \ - --set=GROUP_NAME=\'$1\' --set=METRIC=\'$2\' --set=MONITOR_TYPE=$3 \ +# +# Modify the following psql parameters to suit your own datbase +# +psql -U runcon_admin -h sbnd-db -p 5434 -d sbnd_online_prd \ + --set=GROUP_NAME=\'$1\' --set=METRIC=\'$2\' --set=MONITOR_TYPE=\'$3\' \ --set=CH1=$4 --set=CHN=$5 -c "\i channelMonitor.sql" diff --git a/sbndqm/Archiver/ddl/monitorMap.sql b/sbndqm/Archiver/ddl/monitorMap.sql index 222d65c..8d950d7 100644 --- a/sbndqm/Archiver/ddl/monitorMap.sql +++ b/sbndqm/Archiver/ddl/monitorMap.sql @@ -1,27 +1,32 @@ -CREATE TABLE MONITOR_TYPES +CREATE TABLE RUNCON_PRD.MONITOR_TYPES ( MONITOR_TYPE INTEGER NOT NULL, TYPE_NAME CHARACTER VARYING(32) NOT NULL, PRIMARY KEY (MONITOR_TYPE) ); -INSERT INTO MONITOR_TYPES VALUES (0,'MEAN'); -INSERT INTO MONITOR_TYPES VALUES (1,'MEDIAN'); -INSERT INTO MONITOR_TYPES VALUES (2,'MODE'); -INSERT INTO MONITOR_TYPES VALUES (3,'MAX'); -INSERT INTO MONITOR_TYPES VALUES (4,'MIN'); -INSERT INTO MONITOR_TYPES VALUES (5,'LAST'); +INSERT INTO RUNCON_PRD.MONITOR_TYPES VALUES (0,'MEAN'); +INSERT INTO RUNCON_PRD.MONITOR_TYPES VALUES (1,'MEDIAN'); +INSERT INTO RUNCON_PRD.MONITOR_TYPES VALUES (2,'MODE'); +INSERT INTO RUNCON_PRD.MONITOR_TYPES VALUES (3,'MAX'); +INSERT INTO RUNCON_PRD.MONITOR_TYPES VALUES (4,'MIN'); +INSERT INTO RUNCON_PRD.MONITOR_TYPES VALUES (5,'LAST'); -CREATE TABLE MONITOR_MAP +CREATE TABLE RUNCON_PRD.MONITOR_MAP ( - GROUP_NAME CHARACTER VARYING(128) NOT NULL, - METRIC CHARACTER VARYING(32) NOT NULL, - CHANNEL_ID INTEGER NOT NULL, - MONITOR_TYPE INTEGER NOT NULL REFERENCES MONITOR_TYPES(MONITOR_TYPE), - POSTGRES_TABLE CHARACTER VARYING(128) NOT NULL, - TIME_INTERVAL DOUBLE PRECISION NOT NULL, - CREATE_TIME TIMESTAMP WITH TIME ZONE DEFAULT NOW(), - PRIMARY KEY (GROUP_NAME,METRIC,CHANNEL_ID,MONITOR_TYPE) + GROUP_NAME CHARACTER VARYING(128) NOT NULL, + METRIC CHARACTER VARYING(32) NOT NULL, + CHANNEL_ID INTEGER NOT NULL, + MONITOR_TYPE INTEGER NOT NULL REFERENCES MONITOR_TYPES(MONITOR_TYPE), + POSTGRES_TABLE CHARACTER VARYING(128) NOT NULL, + TIME_INTERVAL DOUBLE PRECISION DEFAULT 3600, + CREATE_TIME TIMESTAMP WITH TIME ZONE DEFAULT NOW(), + LAST_SMPL_TIME_A TIMESTAMP WITH TIME ZONE, + LAST_SMPL_TIME_B TIMESTAMP WITH TIME ZONE, + LAST_SMPL_VALUE_A DOUBLE PRECISION, + LAST_SMPL_VALUE_B DOUBLE PRECISION, + LAST_SMPL CHAR(1), + PRIMARY KEY (POSTGRES_TABLE,CHANNEL_ID) ); From 216f5fb68e59d2b603860098b3458089bd7636be Mon Sep 17 00:00:00 2001 From: Gennadiy Date: Fri, 3 Oct 2025 08:56:02 -0500 Subject: [PATCH 6/9] FIXED: removed '.decode('utf-8')' from L158 --- sbndqm/Archiver/Archiver.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/sbndqm/Archiver/Archiver.py b/sbndqm/Archiver/Archiver.py index b6d194f..4e4b5a1 100644 --- a/sbndqm/Archiver/Archiver.py +++ b/sbndqm/Archiver/Archiver.py @@ -154,7 +154,9 @@ def ProcessStreams(r, p, cur, StreamDict, Config, args): #Loop over the streams which have entries to be archived. for StreamObject in ReadStream: stream_name_b = StreamObject[0] - stream_name = stream_name_b.decode('utf-8') + # print("DEBUG: stream_name_b =", stream_name_b, "type =", type(stream_name_b)) + # stream_name = stream_name_b.decode('utf-8') + stream_name = stream_name_b entries = StreamObject[1] #Loop over the individual entries in the stream. for DataObject in entries: From 4008f84c93110ef312e0701895aa2c117278f23a Mon Sep 17 00:00:00 2001 From: Gennadiy Date: Fri, 3 Oct 2025 10:01:27 -0500 Subject: [PATCH 7/9] Updated script --- sbndqm/Archiver/RestartArchiver.sh | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/sbndqm/Archiver/RestartArchiver.sh b/sbndqm/Archiver/RestartArchiver.sh index 8f2be42..8654c07 100644 --- a/sbndqm/Archiver/RestartArchiver.sh +++ b/sbndqm/Archiver/RestartArchiver.sh @@ -4,7 +4,14 @@ echo "Remaining OnMon Archiver.py processes:" ps aux | grep '[p]ython Archiver.py' echo "Killing remaining OnMon Archiver.py processes..." -kill -9 $(ps aux | grep '[p]ython Archiver.py' | awk '{print $2}') + +toKillProc=$(ps aux | grep '[p]ython Archiver.py' | awk '{print $2}') +if [[ $toKillProc != '' ]] ; then + kill -9 $toKillProc +else + echo "No OnMon Archiver.py processes remaning..." +fi + echo "Restarting OnMon Archiver." cd /home/nfs/icarus/Archiver/sbndqm/sbndqm/Archiver From 2e910b6e1abeec41aceb138e3ac951809c4a2c02 Mon Sep 17 00:00:00 2001 From: Gennadiy Date: Fri, 3 Oct 2025 10:27:00 -0500 Subject: [PATCH 8/9] Added try-catch for some errors on reading data stream --- sbndqm/Archiver/Archiver.py | 16 +++++++++++----- sbndqm/Archiver/RestartArchiver.sh | 2 +- 2 files changed, 12 insertions(+), 6 deletions(-) diff --git a/sbndqm/Archiver/Archiver.py b/sbndqm/Archiver/Archiver.py index 4e4b5a1..843922e 100644 --- a/sbndqm/Archiver/Archiver.py +++ b/sbndqm/Archiver/Archiver.py @@ -128,7 +128,10 @@ def ProcessStreams(r, p, cur, StreamDict, Config, args): #ReadStream is a list of stream object which have new entries that have yet to be archived. try: st = time.time() - ReadStream = r.xread(StreamDict, block=0) + try: + ReadStream = r.xread(StreamDict, block=0) + except Exception as e: + logging.error(e) logging.info('XREAD time: {}'.format(time.time()-st)) #logging.info('Redis READ completed.') except redis.RedisError as err: @@ -153,16 +156,19 @@ def ProcessStreams(r, p, cur, StreamDict, Config, args): #Loop over the streams which have entries to be archived. for StreamObject in ReadStream: - stream_name_b = StreamObject[0] + # stream_name_b = StreamObject[0] # print("DEBUG: stream_name_b =", stream_name_b, "type =", type(stream_name_b)) # stream_name = stream_name_b.decode('utf-8') - stream_name = stream_name_b + stream_name = StreamObject[0] entries = StreamObject[1] #Loop over the individual entries in the stream. for DataObject in entries: entry_id_b, entry_fields = DataObject - entry_id = entry_id_b.decode('utf-8') - + try: + entry_id = entry_id_b.decode('utf-8') + except Exception as e: + logging.error(e) + #Check if the latest completed archived entry is '0' (no metrics have been archived yet for the stream). if StreamDict[stream_name] == '0': NewLatest = str( int( entry_id.split('-')[0] ) - 1 ) + '-0' diff --git a/sbndqm/Archiver/RestartArchiver.sh b/sbndqm/Archiver/RestartArchiver.sh index 8654c07..f667566 100644 --- a/sbndqm/Archiver/RestartArchiver.sh +++ b/sbndqm/Archiver/RestartArchiver.sh @@ -6,7 +6,7 @@ ps aux | grep '[p]ython Archiver.py' echo "Killing remaining OnMon Archiver.py processes..." toKillProc=$(ps aux | grep '[p]ython Archiver.py' | awk '{print $2}') -if [[ $toKillProc != '' ]] ; then +if [[ $toKillProc ]] ; then kill -9 $toKillProc else echo "No OnMon Archiver.py processes remaning..." From 1dead2a82cf5230de186b605c74c6ceb4d52033f Mon Sep 17 00:00:00 2001 From: Gennadiy Date: Fri, 3 Oct 2025 16:31:03 -0500 Subject: [PATCH 9/9] add debugging --- sbndqm/Archiver/Archiver.py | 150 ++++++++++++++++++++---------------- 1 file changed, 85 insertions(+), 65 deletions(-) diff --git a/sbndqm/Archiver/Archiver.py b/sbndqm/Archiver/Archiver.py index 843922e..dbb4b15 100644 --- a/sbndqm/Archiver/Archiver.py +++ b/sbndqm/Archiver/Archiver.py @@ -133,7 +133,7 @@ def ProcessStreams(r, p, cur, StreamDict, Config, args): except Exception as e: logging.error(e) logging.info('XREAD time: {}'.format(time.time()-st)) - #logging.info('Redis READ completed.') + logging.info('Redis READ completed.') except redis.RedisError as err: logging.error('Error while reading streams: {}'.format(err)) logging.error(type(err)) @@ -155,84 +155,99 @@ def ProcessStreams(r, p, cur, StreamDict, Config, args): ReconnectCount = 0 #Loop over the streams which have entries to be archived. + countStreams = 0 + totalStreams = len(ReadStream) + logging.info("Streams to be archived: {}".format(totalStreams)) + for StreamObject in ReadStream: # stream_name_b = StreamObject[0] # print("DEBUG: stream_name_b =", stream_name_b, "type =", type(stream_name_b)) # stream_name = stream_name_b.decode('utf-8') stream_name = StreamObject[0] entries = StreamObject[1] + logging.debug("Stream: {}".format(stream_name)) + #Loop over the individual entries in the stream. for DataObject in entries: entry_id_b, entry_fields = DataObject try: entry_id = entry_id_b.decode('utf-8') - except Exception as e: - logging.error(e) - - #Check if the latest completed archived entry is '0' (no metrics have been archived yet for the stream). - if StreamDict[stream_name] == '0': - NewLatest = str( int( entry_id.split('-')[0] ) - 1 ) + '-0' - TotalSetTime += SetLatest(r, stream_name, NewLatest) - TotalSetN += 1 - ProcessData(MetricDict, (entry_id, entry_fields), Config, stream_name) - StreamDict[stream_name] = entry_id - #Check if the current entry is within the time block (contained in the interval of entries to be archived). - elif int( entry_id.split('-')[0] ) < int( GetLatest(r, stream_name).split('-')[0] ) + 1000*Config[stream_name][5]: - ProcessData(MetricDict, (entry_id, entry_fields), Config, stream_name) - StreamDict[stream_name] = entry_id - #Make sure that there are metrics to archive (No metrics in MetricDict, but GetLatest is more than one time block previous to the next entry). - elif int( entry_id.split('-')[0] ) > int( GetLatest(r, stream_name).split('-')[0] ) + 1000*Config[stream_name][5] and len(MetricDict[stream_name]) == 0: - NewLatest = str( int( entry_id.split('-')[0] ) - 1 ) + '-0' - TotalSetTime += SetLatest(r, stream_name, NewLatest) - TotalSetN += 1 - ProcessData(MetricDict, (entry_id, entry_fields), Config, stream_name) - StreamDict[stream_name] = entry_id - #The entry is outside the current time block. Set a new latest completed and write to the database. - else: - #Case for "mean" averaging. - if Config[stream_name][3] == 0: - TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][0]) - TotalSetN += 1 - WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], MetricDict[stream_name][1], int(MetricDict[stream_name][0].split('-')[0])) - MetricDict[stream_name] = [] - #Case for "median" averaging. - elif Config[stream_name][3] == 1: - TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][-1][0]) - TotalSetN += 1 - MetricList = [ x[1] for x in MetricDict[stream_name] ] - WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], median(MetricList), int(MetricDict[stream_name][-1][0].split('-')[0])) - MetricDict[stream_name] = [] - #Case for "mode" averaging. - elif Config[stream_name][3] == 2: - TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][0]) - TotalSetN += 1 - WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], MetricDict[stream_name][3], int(MetricDict[stream_name][0].split('-')[0])) - MetricDict[stream_name] = [] - #Case for "max" averaging. - elif Config[stream_name][3] == 3: - TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][0]) - TotalSetN += 1 - WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], MetricDict[stream_name][1], int(MetricDict[stream_name][0].split('-')[0])) - MetricDict[stream_name] = [] - #Case for "min" averaging. - elif Config[stream_name][3] == 4: - TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][0]) + + logging.debug("Entry: {}, fields: {}".format(entry_id,entry_fields)) + + #Check if the latest completed archived entry is '0' (no metrics have been archived yet for the stream). + if StreamDict[stream_name] == '0': + NewLatest = str( int( entry_id.split('-')[0] ) - 1 ) + '-0' + TotalSetTime += SetLatest(r, stream_name, NewLatest) TotalSetN += 1 - WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], MetricDict[stream_name][1], int(MetricDict[stream_name][0].split('-')[0])) - MetricDict[stream_name] = [] - #Case for "last" averaging. - elif Config[stream_name][3] == 5: - TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][0]) - WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], MetricDict[stream_name][1], int(MetricDict[stream_name][0].split('-')[0])) - MetricDict[stream_name] = [] - #After performing the archiving on the previous block of data, we can now process the current entry. - ProcessData(MetricDict, (entry_id, entry_fields), Config, stream_name) - StreamDict[stream_name] = entry_id - #Check to see if there is a gap in the data stream (the current object is more than one time interval from the latest completed). - if int( entry_id.split('-')[0] ) > int( GetLatest(r, stream_name).split('-')[0] ) + 1000*Config[stream_name][5]: + ProcessData(MetricDict, (entry_id, entry_fields), Config, stream_name) + StreamDict[stream_name] = entry_id + #Check if the current entry is within the time block (contained in the interval of entries to be archived). + elif int( entry_id.split('-')[0] ) < int( GetLatest(r, stream_name).split('-')[0] ) + 1000*Config[stream_name][5]: + ProcessData(MetricDict, (entry_id, entry_fields), Config, stream_name) + StreamDict[stream_name] = entry_id + #Make sure that there are metrics to archive (No metrics in MetricDict, but GetLatest is more than one time block previous to the next entry). + elif int( entry_id.split('-')[0] ) > int( GetLatest(r, stream_name).split('-')[0] ) + 1000*Config[stream_name][5] and len(MetricDict[stream_name]) == 0: NewLatest = str( int( entry_id.split('-')[0] ) - 1 ) + '-0' TotalSetTime += SetLatest(r, stream_name, NewLatest) TotalSetN += 1 + ProcessData(MetricDict, (entry_id, entry_fields), Config, stream_name) + StreamDict[stream_name] = entry_id + #The entry is outside the current time block. Set a new latest completed and write to the database. + else: + #Case for "mean" averaging. + if Config[stream_name][3] == 0: + TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][0]) + TotalSetN += 1 + WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], MetricDict[stream_name][1], int(MetricDict[stream_name][0].split('-')[0])) + MetricDict[stream_name] = [] + #Case for "median" averaging. + elif Config[stream_name][3] == 1: + TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][-1][0]) + TotalSetN += 1 + MetricList = [ x[1] for x in MetricDict[stream_name] ] + WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], median(MetricList), int(MetricDict[stream_name][-1][0].split('-')[0])) + MetricDict[stream_name] = [] + #Case for "mode" averaging. + elif Config[stream_name][3] == 2: + TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][0]) + TotalSetN += 1 + WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], MetricDict[stream_name][3], int(MetricDict[stream_name][0].split('-')[0])) + MetricDict[stream_name] = [] + #Case for "max" averaging. + elif Config[stream_name][3] == 3: + TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][0]) + TotalSetN += 1 + WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], MetricDict[stream_name][1], int(MetricDict[stream_name][0].split('-')[0])) + MetricDict[stream_name] = [] + #Case for "min" averaging. + elif Config[stream_name][3] == 4: + TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][0]) + TotalSetN += 1 + WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], MetricDict[stream_name][1], int(MetricDict[stream_name][0].split('-')[0])) + MetricDict[stream_name] = [] + #Case for "last" averaging. + elif Config[stream_name][3] == 5: + TotalSetTime += SetLatest(r, stream_name, MetricDict[stream_name][0]) + WritePostgres(p, cur, Config[stream_name][4], Config[stream_name][2], MetricDict[stream_name][1], int(MetricDict[stream_name][0].split('-')[0])) + MetricDict[stream_name] = [] + #After performing the archiving on the previous block of data, we can now process the current entry. + ProcessData(MetricDict, (entry_id, entry_fields), Config, stream_name) + StreamDict[stream_name] = entry_id + #Check to see if there is a gap in the data stream (the current object is more than one time interval from the latest completed). + if int( entry_id.split('-')[0] ) > int( GetLatest(r, stream_name).split('-')[0] ) + 1000*Config[stream_name][5]: + NewLatest = str( int( entry_id.split('-')[0] ) - 1 ) + '-0' + TotalSetTime += SetLatest(r, stream_name, NewLatest) + TotalSetN += 1 + + except Exception as e: + logging.exception(e) + + if countStreams % 5000 == 0 or countStreams == totalStreams: + percentDone = (countStreams / totalStreams) * 100 + logging.info("Processed {}/{} streams ({:.2f}%)".format(countStreams,totalStreams,percentDone)) + countStreams += 1 + logging.info('Total set time: {}'.format(TotalSetTime)) logging.info('Average set time: {}'.format(float(TotalSetTime)/float(TotalSetN) if TotalSetN else 0.0)) @@ -415,4 +430,9 @@ def signal_handler(sig, frame): parser = argparse.ArgumentParser() parser.add_argument("-pw", "--password", default=None) parser.add_argument("-pr", "--processes", default=1) - main(parser.parse_args()) + try: + main(parser.parse_args()) + except Exception as e: + logging.exception("Uncaught exception in main()") + logging.exception(f"Archiver crashed with exception: {e}") + sys.exit(1)