pandaharvester/harvestercredmanager/superfacility_cred_manager.py

0001 import os
0002 from datetime import datetime, timedelta
0003
0004 from pandaharvester.harvestercore import core_utils
0005
0006 from .base_cred_manager import BaseCredManager
0007
0008 # logger
0009 _logger = core_utils.setup_logger("superfacility_cred_manager")
0010
0011
0012 class SuperfacilityCredManager(BaseCredManager):
0013     # constructor
0014     def __init__(self, **kwarg):
0015         BaseCredManager.__init__(self, **kwarg)
0016         # make logger
0017         main_log = self.make_logger(_logger, method_name="__init__")
0018         # set up with direct attributes
0019         self.setupMap = dict(vars(self))
0020         # setupMap
0021         self.cred_dir = self.setupMap.get("cred_dir")
0022         self.lifetime_days = int(self.setupMap.get("lifetime_days", 30))
0023         self.date_format = "%Y-%m-%d-%H%M"
0024
0025     # check proxy lifetime for monitoring/alerting purposes
0026     def check_credential_lifetime(self):
0027         main_log = self.make_logger(_logger, method_name="check_credential_lifetime")
0028         lifetime = None
0029         try:
0030             real_dir = os.path.realpath(self.cred_dir)
0031             basename = os.path.basename(real_dir)
0032             created_dt = datetime.strptime(basename, self.date_format)
0033             age = datetime.now() - created_dt
0034             lifetime_all = timedelta(days=self.lifetime_days)
0035             main_log.debug(f"age={age}, lifetime_all={lifetime_all}")
0036             timediff = lifetime_all - age
0037             lifetime = timediff.days * 24 + timediff.seconds / 3600
0038         except Exception:
0039             core_utils.dump_error_message(main_log)
0040         if isinstance(lifetime, float):
0041             main_log.debug(f"returning lifetime (in hour) {lifetime:.3f}")
0042         else:
0043             main_log.debug(f"returning lifetime (in hour) {lifetime}")
0044         return lifetime
0045
0046     def check_credential(self):
0047         # make logger
0048         main_log = self.make_logger(_logger, method_name="check_credential")
0049         try:
0050             real_dir = os.path.realpath(self.cred_dir)
0051             basename = os.path.basename(real_dir)
0052             created_dt = datetime.strptime(basename, self.date_format)
0053             age = datetime.now() - created_dt
0054             lifetime_all = timedelta(days=self.lifetime_days)
0055             main_log.debug(f"age={age}, lifetime_all={lifetime_all}")
0056             return age <= lifetime_all
0057         except Exception as e:
0058             main_log.debug(f"[SuperfacilityCredManager] check_credential func call with error: {e}")
0059             core_utils.dump_error_message(main_log)
0060             return False
0061
0062     def renew_credential(self):
0063         # Does not support automatic cred updates
0064         # If this method is called, should return error immediately
0065         real_dir = os.path.realpath(self.cred_dir)
0066         msg = f"SuperfacilityClient credentials expired or invalid: symlink {self.cred_dir} points to {real_dir} which is older than {self.lifetime_days} days. "
0067         return False, msg