pandajedi/jediorder/WatchDog.py

0001 import datetime
0002 import os
0003 import socket
0004 import sys
0005 import time
0006
0007 from pandacommon.pandalogger.PandaLogger import PandaLogger
0008 from pandacommon.pandautils.PandaUtils import naive_utcnow
0009
0010 from pandajedi.jediconfig import jedi_config
0011 from pandajedi.jedicore import Interaction
0012 from pandajedi.jedicore.FactoryBase import FactoryBase
0013 from pandajedi.jedicore.MsgWrapper import MsgWrapper
0014
0015 from .JediKnight import JediKnight
0016
0017 logger = PandaLogger().getLogger(__name__.split(".")[-1])
0018
0019
0020 # worker class for watchdog
0021 class WatchDog(JediKnight, FactoryBase):
0022     # constructor
0023     def __init__(self, commuChannel, taskBufferIF, ddmIF, vos, prodSourceLabels, subStr, period):
0024         self.vos = self.parseInit(vos)
0025         self.prodSourceLabels = self.parseInit(prodSourceLabels)
0026         self.subStr = subStr
0027         self.period = period
0028         self.pid = f"{socket.getfqdn().split('.')[0]}-{os.getpid()}-dog"
0029         JediKnight.__init__(self, commuChannel, taskBufferIF, ddmIF, logger)
0030         FactoryBase.__init__(self, self.vos, self.prodSourceLabels, logger, jedi_config.watchdog.modConfig)
0031
0032     # main
0033     def start(self):
0034         # start base classes
0035         JediKnight.start(self)
0036         FactoryBase.initializeMods(self, self.taskBufferIF, self.ddmIF)
0037         # go into main loop
0038         while True:
0039             startTime = naive_utcnow()
0040             try:
0041                 # get logger
0042                 tmpLog = MsgWrapper(logger)
0043                 tmpLog.info("start")
0044                 # loop over all vos
0045                 for vo in self.vos:
0046                     # loop over all sourceLabels
0047                     for prodSourceLabel in self.prodSourceLabels:
0048                         # vo/prodSourceLabel specific action
0049                         impl = self.getImpl(vo, prodSourceLabel, subType=self.subStr)
0050                         if impl is not None:
0051                             plugin_name = impl.__class__.__name__
0052                             tmpLog.info(f"pre-action for vo={vo} label={prodSourceLabel} cls={plugin_name}")
0053                             impl.pre_action(tmpLog, vo, prodSourceLabel, self.pid)
0054                             tmpLog.info(f"do action for vo={vo} label={prodSourceLabel} cls={plugin_name}")
0055                             tmpStat = impl.doAction()
0056                             if tmpStat != Interaction.SC_SUCCEEDED:
0057                                 tmpLog.error(f"failed to run special action for vo={vo} label={prodSourceLabel} cls={plugin_name}")
0058                             else:
0059                                 tmpLog.info(f"done for vo={vo} label={prodSourceLabel} cls={plugin_name}")
0060                 tmpLog.info("done")
0061             except Exception:
0062                 errtype, errvalue = sys.exc_info()[:2]
0063                 tmpLog.error(f"failed in {self.__class__.__name__}.start() with {errtype.__name__} {errvalue}")
0064             # sleep if needed
0065             loopCycle = jedi_config.watchdog.loopCycle if self.period is None else self.period
0066             timeDelta = naive_utcnow() - startTime
0067             sleepPeriod = loopCycle - timeDelta.seconds
0068             if sleepPeriod > 0:
0069                 time.sleep(sleepPeriod)
0070             # randomize cycle
0071             self.randomSleep(max_val=loopCycle)
0072
0073
0074 # launch
0075
0076
0077 def launcher(commuChannel, taskBufferIF, ddmIF, vos=None, prodSourceLabels=None, subStr=None, period=None):
0078     p = WatchDog(commuChannel, taskBufferIF, ddmIF, vos, prodSourceLabels, subStr, period)
0079     p.start()