pandaharvester/harvestercredmanager/iam_token_cred_manager_globus_compute.py

0001 import os
0002 import traceback
0003
0004 from pandaharvester.harvestercore import core_utils
0005 from pandaharvester.harvestermisc.token_utils import endpoint_to_filename
0006 from pandaharvester.harvestermisc.globus_compute_token_sync import GlobusComputeTokenReplicator
0007
0008 from .iam_token_cred_manager import IamTokenCredManager
0009
0010 _logger = core_utils.setup_logger("iam_token_cred_manager_gc")
0011
0012 class IamTokenCredManagerRemoteGlobusCompute(IamTokenCredManager):
0013     def __init__(self, **kwarg):
0014         super().__init__(**kwarg)
0015         tmp_log = self.make_logger(_logger, f"config={self.setup_name}", method_name="__init__")
0016
0017         self.remote_out_dir = self.setupMap.get("remote_out_dir", "")
0018         gc_cfg = self.setupMap.get("globus_compute", {})
0019         if gc_cfg and self.remote_out_dir:
0020             self.replicator = GlobusComputeTokenReplicator(gc_cfg, tmp_log)
0021         else:
0022             tmp_log.debug(f"replicator is not initialized as either gc_cfg or remote_out_dir is missing. They are gc_cfg = {gc_cfg} and remote_out_dir = {self.remote_out_dir}")
0023             self.replicator = None
0024
0025     # If not specify remote token replicator, then do nothing
0026     # First get the final remote token filepath
0027     # Then inside token replicator, encrypt the token locally as a string
0028     # Next within token replicator submit a task to transfer the token to a tmp place
0029     # Inside that task, next decrypt the token and atomically replace the old remote token file
0030     def _sync_remote(self, token_filename: str, token_str: str, logger):
0031         if not self.replicator:
0032             return
0033         remote_token_path = os.path.join(self.remote_out_dir, token_filename)
0034         status = self.replicator.do_it(token_str, remote_token_path)
0035         if status:
0036             logger.info(f"Synchronized token '{token_filename}' to remote: {remote_token_path}")
0037         else:
0038             logger.error(f"FAILED to synchronize token '{token_filename}' to remote: {remote_token_path}")
0039
0040     def renew_credential(self):
0041         # make logger
0042         tmp_log = self.make_logger(_logger, f"config={self.setup_name}", method_name="renew_credential")
0043         # go
0044         all_ok = True
0045         all_err_str = ""
0046         for target in self.targets_dict:
0047             try:
0048                 # write to file
0049                 if self.target_type == "panda":
0050                     token_filename = self.panda_token_filename
0051                 else:
0052                     token_filename = endpoint_to_filename(target)
0053                 token_path = os.path.join(self.out_dir, token_filename)
0054                 # check token freshness
0055                 if self._is_fresh(token_path):
0056                     # token still fresh, skip it
0057                     tmp_log.debug(f"token for {target} at {token_path} still fresh; skipped")
0058                     continue
0059                 # renew access token of target locally
0060                 access_token = self.issuer_broker.get_access_token(aud=target, scope=self.scope)
0061                 with open(token_path, "w") as f:
0062                     f.write(access_token)
0063                 tmp_log.info(f"renewed token for {target} at {token_path}")
0064                 # encrypt, and sync to remote (if configured) per token file mapping, and then decrypt remotely
0065                 try:
0066                     self._sync_remote(token_filename, access_token, tmp_log)
0067                 except Exception:
0068                     all_ok = False
0069                     tmp_log.error(f"Remote sync failed for {target}. {traceback.format_exc()}")
0070                     all_err_str = "failed to sync some tokens; see plugin log for details "
0071
0072             except Exception as e:
0073                 err_str = f"Problem getting token for {target}. {traceback.format_exc()}"
0074                 tmp_log.error(err_str)
0075                 all_ok = False
0076                 all_err_str = "failed to get some tokens. Check the plugin log for details "
0077                 continue
0078         # update last timestamp
0079         self._update_ts()
0080         tmp_log.debug("done")
0081         # return
0082         return all_ok, all_err_str