Merge pull request #218 from wguanicedew/dev

improve

client/bin/idds

@@ -6,7 +6,7 @@
 # http://www.apache.org/licenses/LICENSE-2.0OA
 #
 # Authors:
-# - Wen Guan, <[email protected]>, 2020 - 2022
+# - Wen Guan, <[email protected]>, 2020 - 2023
 
 """
 iDDS CLI
@@ -42,6 +42,11 @@ def setup_oidc_token(args):
     cm = ClientManager(host=args.host, setup_client=True)
     cm.setup_oidc_token()
 
+def setup_oidc_client_token(args):
+    cm = ClientManager(host=args.host, setup_client=True)
+    cm.setup_oidc_client_token(issuer=args.oidc_issuer, client_id=args.oidc_client_id,
+                               client_secret=args.oidc_client_secret, scope=args.oidc_scope,
+                               audience=args.oidc_audience)
 
 def clean_oidc_token(args):
     cm = ClientManager(host=args.host, setup_client=False)
@@ -151,6 +156,9 @@ def get_parser():
     # setup token
     token_setup_parser = subparsers.add_parser('setup_oidc_token', help='Setup authentication token')
     token_setup_parser.set_defaults(function=setup_oidc_token)
+    # token_setup_parser.add_argument('--oidc_audience', dest='oidc_audience', default=None, help='The oidc audience')
+    # token_setup_parser.add_argument('--oidc_scope', dest='oidc_scope', default=None, help='The oidc scope')
+
     # token_setup_parser.add_argument('--oidc_refresh_lifetime', dest='oidc_refresh_lifetime', default=None, help='The oidc refresh lifetime')
     # token_setup_parser.add_argument('--oidc_issuer', dest='oidc_issuer', default=None, help='The oidc issuer')
     # token_setup_parser.add_argument('--oidc_audience', dest='oidc_audience', default=None, help='The oidc audience')
@@ -163,6 +171,14 @@ def get_parser():
     # token_setup_parser.add_argument('--saml_username', dest='saml_username', default=None, help='The SAML username')
     # token_setup_parser.add_argument('--saml_password', dest='saml_password', default=None, help='The saml password')
 
+    token_setup_parser = subparsers.add_parser('setup_oidc_client_token', help='Setup authentication client token')
+    token_setup_parser.set_defaults(function=setup_oidc_client_token)
+    token_setup_parser.add_argument('--oidc_issuer', dest='oidc_issuer', default=None, help='The oidc issuer')
+    token_setup_parser.add_argument('--oidc_client_id', dest='oidc_client_id', default=None, help='The oidc client id')
+    token_setup_parser.add_argument('--oidc_client_secret', dest='oidc_client_secret', default=None, help='The oidc client secret')
+    token_setup_parser.add_argument('--oidc_audience', dest='oidc_audience', default=None, help='The oidc audience')
+    token_setup_parser.add_argument('--oidc_scope', dest='oidc_scope', default=None, help='The oidc scope')
+
     # clean token
     token_clean_parser = subparsers.add_parser('clean_oidc_token', help='Clean authentication token')
     token_clean_parser.set_defaults(function=clean_oidc_token)

client/lib/idds/client/clientmanager.py

@@ -31,7 +31,7 @@
     raw_input = input
 
 
-from idds.common.authentication import OIDCAuthenticationUtils
+from idds.common.authentication import OIDCAuthentication, OIDCAuthenticationUtils
 from idds.common.utils import setup_logging, get_proxy_path
 
 from idds.client.version import release_version
@@ -316,6 +316,27 @@ def setup_oidc_token(self):
             else:
                 logging.info("Failed to save token to %s: (status: %s, output: %s)" % (self.oidc_token_file, status, output))
 
+    def setup_oidc_client_token(self, issuer, client_id, client_secret, scope, audience):
+        """"
+        Setup oidc client token
+        """
+        self.setup_client(auth_setup=True)
+        oidc_auth = OIDCAuthentication()
+        status, token = oidc_auth.setup_oidc_client_token(issuer=issuer, client_id=client_id,
+                                                          client_secret=client_secret, scope=scope,
+                                                          audience=audience)
+
+        if not status:
+            logging.error("Failed to get token.")
+        else:
+            oidc_util = OIDCAuthenticationUtils()
+            token_file = self.oidc_token_file + "_client"
+            status, output = oidc_util.save_token(token_file, token)
+            if status:
+                logging.info("Token is saved to %s" % (token_file))
+            else:
+                logging.info("Failed to save token to %s: (status: %s, output: %s)" % (token_file, status, output))
+
     def refresh_oidc_token(self):
         """"
         refresh oidc token

common/lib/idds/common/authentication.py

@@ -338,6 +338,41 @@ def verify_id_token(self, vo, token):
         except Exception as error:
             return False, 'Failed to verify oidc token: ' + str(error), None
 
+    def setup_oidc_client_token(self, issuer, client_id, client_secret, scope, audience):
+        try:
+            data = {'client_id': client_id,
+                    'client_secret': client_secret,
+                    'grant_type': 'client_credentials',
+                    'scope': scope,
+                    'audience': audience}
+
+            headers = {'content-type': 'application/x-www-form-urlencoded'}
+
+            endpoint = '{0}/token'.format(issuer)
+            result = requests.session().post(endpoint,
+                                             # data=json.dumps(data),
+                                             # data=data,
+                                             urlencode(data).encode(),
+                                             timeout=self.timeout,
+                                             verify=should_verify(ssl_verify=self.get_ssl_verify()),
+                                             headers=headers)
+
+            if result is not None:
+                # print(result)
+                # print(result.text)
+                # print(result.status_code)
+
+                if result.status_code == HTTP_STATUS_CODE.OK and result.text:
+                    return True, json.loads(result.text)
+                else:
+                    return False, "Failed to refresh oidc token (status: %s, text: %s)" % (result.status_code, result.text)
+            else:
+                return False, "Failed to refresh oidc token. Response is None."
+        except requests.exceptions.ConnectionError as error:
+            return False, 'Failed to refresh oidc token. ConnectionError: ' + str(error)
+        except Exception as error:
+            return False, 'Failed to refresh oidc token: ' + str(error)
+
 
 class OIDCAuthenticationUtils(object):
     def __init__(self):

doma/lib/idds/doma/workflowv2/domapandawork.py

@@ -883,6 +883,9 @@ def get_update_contents(self, unterminated_jobs_status, input_output_maps, conte
             panda_status = panda_job_status['status']
             job_info = panda_job_status['job_info']
 
+            if input_file not in inputname_to_map_id_outputs:
+                continue
+
             output_contents = inputname_to_map_id_outputs[input_file]['outputs']
             for content in output_contents:
                 content['substatus'] = panda_status

main/etc/sql/oracle_update.sql

@@ -446,8 +446,8 @@ alter table contents add constraint CONTENT_ID_UQ UNIQUE (transform_id, coll_id,
 
 --- 20230927
 alter table contents add (name_md5 varchar2(33), scope_name_md5 varchar2(33));
-update table contents set name_md5=md5(name), scope_name_md5=md5(scope || name);
+update contents set name_md5=standard_hash(name, 'MD5'), scope_name_md5=standard_hash(scope || name, 'MD5');
 alter table contents drop constraint CONTENT_ID_UQ;
 alter table contents add constraint CONTENT_ID_UQ UNIQUE (transform_id, coll_id, map_id, sub_map_id, dep_sub_map_id, content_relation_type, name_md5, scope_name_md5, min_id, max_id) USING INDEX LOCAL;
 drop index CONTENTS_ID_NAME_IDX;
-CREATE INDEX CONTENTS_ID_NAME_IDX ON CONTENTS (coll_id, scope, md5(name), status);
+CREATE INDEX CONTENTS_ID_NAME_IDX ON CONTENTS (coll_id, scope, standard_hash(name, 'MD5'), status);

main/lib/idds/agents/carrier/utils.py

@@ -1665,30 +1665,39 @@ def sync_collection_status(request_id, transform_id, workload_id, work, input_ou
     return update_collections, all_updates_flushed, messages
 
 
-def sync_work_status(request_id, transform_id, workload_id, work):
+def sync_work_status(request_id, transform_id, workload_id, work, substatus=None):
     input_collections = work.get_input_collections()
     output_collections = work.get_output_collections()
     log_collections = work.get_log_collections()
 
     is_all_collections_closed = True
     is_all_files_processed = True
     is_all_files_failed = True
+    has_files = False
     for coll in input_collections + output_collections + log_collections:
         if coll.status != CollectionStatus.Closed:
             is_all_collections_closed = False
     for coll in output_collections:
+        if coll.total_files > 0:
+            has_files = True
         if coll.total_files != coll.processed_files:
             is_all_files_processed = False
         if coll.processed_files > 0 or coll.total_files == coll.processed_files:
             is_all_files_failed = False
 
     if is_all_collections_closed:
-        if is_all_files_processed:
-            work.status = WorkStatus.Finished
-        elif is_all_files_failed:
-            work.status = WorkStatus.Failed
+        if has_files:
+            if is_all_files_processed:
+                work.status = WorkStatus.Finished
+            elif is_all_files_failed:
+                work.status = WorkStatus.Failed
+            else:
+                work.status = WorkStatus.SubFinished
         else:
-            work.status = WorkStatus.SubFinished
+            if substatus:
+                work.status = substatus
+            else:
+                work.status = WorkStatus.Failed
 
 
 def sync_processing(processing, agent_attributes, terminate=False, abort=False, logger=None, log_prefix=""):
@@ -1715,7 +1724,7 @@ def sync_processing(processing, agent_attributes, terminate=False, abort=False,
 
     messages += msgs
 
-    sync_work_status(request_id, transform_id, workload_id, work)
+    sync_work_status(request_id, transform_id, workload_id, work, processing['substatus'])
     logger.info(log_prefix + "sync_processing: work status: %s" % work.get_status())
     if terminate and work.is_terminated():
         msgs = generate_messages(request_id, transform_id, workload_id, work, msg_type='work')

main/lib/idds/tests/test_domapanda.py

@@ -222,7 +222,7 @@ def setup_workflow():
     work4 = DomaPanDAWork(executable='echo',
                           primary_input_collection={'scope': 'pseudo_dataset', 'name': 'pseudo_input_collection#1'},
                           output_collections=[{'scope': 'pseudo_dataset', 'name': 'pseudo_output_collection#1'}],
-                          log_collections=[], dependency_map=taskN1.dependencies,
+                          log_collections=[], dependency_map=taskN4.dependencies,
                           task_name=taskN4.name, task_queue=task_queue3,
                           encode_command_line=True,
                           task_priority=981,
@@ -238,7 +238,7 @@ def setup_workflow():
     work5 = DomaPanDAWork(executable='echo',
                           primary_input_collection={'scope': 'pseudo_dataset', 'name': 'pseudo_input_collection#1'},
                           output_collections=[{'scope': 'pseudo_dataset', 'name': 'pseudo_output_collection#1'}],
-                          log_collections=[], dependency_map=taskN1.dependencies,
+                          log_collections=[], dependency_map=taskN5.dependencies,
                           task_name=taskN5.name, task_queue=task_queue4,
                           encode_command_line=True,
                           task_priority=981,

main/tools/env/setup_dev.sh

@@ -25,6 +25,8 @@ source /afs/cern.ch/user/w/wguan/workdisk/conda/setup.sh
 conda activate $CondaDir
 #export PYTHONPATH=${IDDS_HOME}/lib:$PYTHONPATH
 
+export IDDS_MAX_NAME_LENGTH=8000
+
 export RUCIO_HOME=$RootDir
 #export RUCIO_ACCOUNT=ddmadmin
 export RUCIO_ACCOUNT=wguan

monitor/data/conf.js

@@ -1,9 +1,9 @@
 
 var appConfig = {
-    'iddsAPI_request': "https://lxplus812.cern.ch:443/idds/monitor_request/null/null",
-    'iddsAPI_transform': "https://lxplus812.cern.ch:443/idds/monitor_transform/null/null",
-    'iddsAPI_processing': "https://lxplus812.cern.ch:443/idds/monitor_processing/null/null",
-    'iddsAPI_request_detail': "https://lxplus812.cern.ch:443/idds/monitor/null/null/true/false/false",
-    'iddsAPI_transform_detail': "https://lxplus812.cern.ch:443/idds/monitor/null/null/false/true/false",
-    'iddsAPI_processing_detail': "https://lxplus812.cern.ch:443/idds/monitor/null/null/false/false/true"
+    'iddsAPI_request': "https://lxplus801.cern.ch:443/idds/monitor_request/null/null",
+    'iddsAPI_transform': "https://lxplus801.cern.ch:443/idds/monitor_transform/null/null",
+    'iddsAPI_processing': "https://lxplus801.cern.ch:443/idds/monitor_processing/null/null",
+    'iddsAPI_request_detail': "https://lxplus801.cern.ch:443/idds/monitor/null/null/true/false/false",
+    'iddsAPI_transform_detail': "https://lxplus801.cern.ch:443/idds/monitor/null/null/false/true/false",
+    'iddsAPI_processing_detail': "https://lxplus801.cern.ch:443/idds/monitor/null/null/false/false/true"
 }