update icarus archiver to python3 by mvicenzi · Pull Request #53 · SBNSoftware/sbndqm

mvicenzi · 2025-09-22T20:42:37Z

This PR does the following:

commits long-standing Archiver.py code developed by @justinjmueller
commits python3 conversion of Archiver.py done by @kyjung123
additional debugging and fixes by @mattiasotgia

mattiasotgia

I left some suggestion on where some try..catch blocks might help
However, the top try:...catch block in the __main__ code should catch all the exceptions thrown by the program

mattiasotgia · 2025-10-06T11:43:34Z

+def LoadConfig(Name):
+    with open(Name) as JSONFile:


I suggest also in this place to add a try...except to catch any failure in reading

try: with open(Name) as JSONFile: Config = json.load(JSONFile) except Exeption as e: loggin.error('Failed to load Archiver configuration: {}'.format(e)) print('Failed to load Archiver configuration: {}'.format(e)) sys.exit(0)

mattiasotgia · 2025-10-06T12:34:55Z

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


Maybe add here a try...except to catch any exception starting process

try: Process(target=Monitor, args=(r, Config)).start() except Exception as e: print('WARNING: Process not started: {}'.format(e)) logging.warning('Process not started: {}'.format(e))

mattiasotgia · 2025-10-06T12:36:43Z

+        ProcessList.append(Process(target=ProcessStreams, args=(r, p, cur, stream_last, per_stream_cfg, args)))
+
    for i in range(ProcessCount):
        ProcessList[i].start()


Same as above, add a try...except to catch exceptions

try: ProcessList[i].start() except Exception as e: print('WARNING: Process not started: {}'.format(e)) logging.warning('Process not started: {}'.format(e))

mattiasotgia · 2025-10-06T13:11:33Z

+            if countStreams % 5000 == 0 or countStreams == totalStreams:
+                percentDone = (countStreams / totalStreams) * 100
+                logging.info("Processed {}/{} streams ({:.2f}%)".format(countStreams,totalStreams,percentDone))
+            countStreams += 1


Maybe add in the archiver configuration the option to select the level of detail of the progress

if countStreams % ArcConfig['logging'].get('print_every', 5000) == 0 or countStreams == totalStreams:

The default (if not present in the configuration json file) is 5000

mattiasotgia · 2025-10-06T13:14:21Z

-        return ParseBinary(val,key)[0]
+        if key == b'val' or key == b'dat': 
+            return val 
+        type_name = key.decode('utf-8')


Add try...except here

try: type_name = key.decode('utf-8') except Exception as e: print('WARNING: decoding error: {}'.format(e)) logging.warning('Decoding error: {}'.format(e))

mattiasotgia · 2025-10-06T13:35:28Z

+
    MetricDict = { x : [] for x in StreamDict.keys()}

    logging.info('Archiver in ProcessStreams.')
+    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.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()
+            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:
+            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(ArcConfig, args)
+                continue
+
+        if len(ReadStream) == 0:
+            logging.error('Block timeout. Reconnecting to Redis...')
+            ArcConfig = LoadConfig('ArchiverConfig.json')
+            r = ConnectRedis(ArcConfig, args)
+            continue
+
+        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 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'
-                    SetLatest(r, StreamObject[0], NewLatest)
-                    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]:
-                    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])
-                        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])
-                        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]))
-                        MetricDict[StreamObject[0]] = []
-                    #Case for "mode" averaging.
-                    elif Config[StreamObject[0]][3] == 2:
-                        SetLatest(r, StreamObject[0], MetricDict[StreamObject[0]][0])
-                        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])
-                        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])
-                        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])
-                        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]:
-                        NewLatest = str( int( DataObject[0].split('-')[0] ) - 1 ) + '-0'
-                        SetLatest(r, StreamObject[0], NewLatest)
+            for DataObject in entries:
+                entry_id_b, entry_fields = DataObject
+                try: 
+                    entry_id = entry_id_b.decode('utf-8')
+
+                    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
+                        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))


Put all the content of the ProcessStreams inside a try...catch block

mattiasotgia · 2025-10-06T13:37:54Z

    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)
-            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()]
+        #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.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)


Put all the content of the Monitor inside of a try...except block

mvicenzi added 6 commits August 8, 2024 21:29

bank Justin's archiver work

e1a1760

remove test scripts, fix indentation

e5a7a85

remove attempted rewrite

aaca5a8

Merge branch 'develop' into feature/icarus_archiver

658752e

python3 conversion by Kiyoung

debff49

restore newer ddl files"

8c63c63

mvicenzi self-assigned this Sep 22, 2025

SilentSegfault added 4 commits October 3, 2025 08:56

FIXED: removed '.decode('utf-8')' from L158

216f5fb

Updated script

4008f84

Added try-catch for some errors on reading data stream

2e910b6

add debugging

1dead2a

mvicenzi changed the base branch from develop to release/v1_05_00 October 4, 2025 03:39

mvicenzi requested a review from mattiasotgia October 4, 2025 03:39

mattiasotgia reviewed Oct 6, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

update icarus archiver to python3#53

update icarus archiver to python3#53
mvicenzi wants to merge 10 commits into
release/v1_05_00from
feature/icarus_archiver

mvicenzi commented Sep 22, 2025 •

edited

Loading

Uh oh!

mattiasotgia left a comment

Uh oh!

mattiasotgia Oct 6, 2025

Uh oh!

mattiasotgia Oct 6, 2025 •

edited

Loading

Uh oh!

mattiasotgia Oct 6, 2025 •

edited

Loading

Uh oh!

mattiasotgia Oct 6, 2025

Uh oh!

mattiasotgia Oct 6, 2025

Uh oh!

mattiasotgia Oct 6, 2025

Uh oh!

mattiasotgia Oct 6, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Uh oh!

Conversation

mvicenzi commented Sep 22, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mattiasotgia left a comment

Choose a reason for hiding this comment

Uh oh!

mattiasotgia Oct 6, 2025

Choose a reason for hiding this comment

Uh oh!

mattiasotgia Oct 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mattiasotgia Oct 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mattiasotgia Oct 6, 2025

Choose a reason for hiding this comment

Uh oh!

mattiasotgia Oct 6, 2025

Choose a reason for hiding this comment

Uh oh!

mattiasotgia Oct 6, 2025

Choose a reason for hiding this comment

Uh oh!

mattiasotgia Oct 6, 2025

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

mvicenzi commented Sep 22, 2025 •

edited

Loading

mattiasotgia Oct 6, 2025 •

edited

Loading

mattiasotgia Oct 6, 2025 •

edited

Loading