<commit_before>import logging
import json
import globus_sdk
import os
import parsl
import typeguard

from functools import partial
from typing import Optional
from parsl.app.app import python_app
from parsl.utils import RepresentationMixin
from parsl.data_provider.staging import Staging

logger = logging.getLogger(__name__)

# globus staging must be run explicitly in the same process/interpreter
# as the DFK. it relies on persisting globus state between actions in that
# process.

"""
'Parsl Application' OAuth2 client registered with Globus Auth
by <EMAIL>
"""
CLIENT_ID = '8b8060fd-610e-4a74-885e-1051c71ad473'
REDIRECT_URI = 'https://auth.globus.org/v2/web/auth-code'
SCOPES = ('openid '
          'urn:globus:auth:scope:transfer.api.globus.org:all')


get_input = getattr(__builtins__, 'raw_input', input)


def _get_globus_provider(dfk, executor_label):
    if executor_label is None:
        raise ValueError("executor_label is mandatory")
    executor = dfk.executors[executor_label]
    if not hasattr(executor, "storage_access"):
        raise ValueError("specified executor does not have storage_access attribute")
    for provider in executor.storage_access:
        if isinstance(provider, GlobusStaging):
            return provider

    raise Exception('No suitable Globus endpoint defined for executor {}'.format(executor_label))


def get_globus():
    Globus.init()
    return Globus()


class Globus(object):
    """
    All communication with the Globus Auth and Globus Transfer services is enclosed
    in the Globus class. In particular, the Globus class is reponsible for:
     - managing an OAuth2 authorizer - getting access and refresh tokens,
       refreshing an access token, storing to and retrieving tokens from
       .globus.json file,
     - submitting file transfers,
     - monitoring transfers.
    """

    authorizer = None

    @classmethod
    def init(cls):
        token_path = os.path.join(os.path.expanduser('~'), '.parsl')
        if not os.path.isdir(token_path):
            os.mkdir(token_path)
        cls.TOKEN_FILE = os.path.join(token_path, '.globus.json')

        if cls.authorizer:
            return
        cls.authorizer = cls._get_native_app_authorizer(CLIENT_ID)

    @classmethod
    def get_authorizer(cls):
        return cls.authorizer

    @classmethod
    def transfer_file(cls, src_ep, dst_ep, src_path, dst_path):
        tc = globus_sdk.TransferClient(authorizer=cls.authorizer)
        td = globus_sdk.TransferData(tc, src_ep, dst_ep)
        td.add_item(src_path, dst_path)
        try:
            task = tc.submit_transfer(td)
        except Exception as e:
            raise Exception('Globus transfer from {}{} to {}{} failed due to error: {}'.format(
                src_ep, src_path, dst_ep, dst_path, e))

        last_event_time = None
        """
        A Globus transfer job (task) can be in one of the three states: ACTIVE, SUCCEEDED, FAILED.
        Parsl every 15 seconds polls a status of the transfer job (task) from the Globus Transfer service,
        with 60 second timeout limit. If the task is ACTIVE after time runs out 'task_wait' returns False,
        and True otherwise.
        """
        while not tc.task_wait(task['task_id'], 60, 15):
            task = tc.get_task(task['task_id'])
            # Get the last error Globus event
            events = tc.task_event_list(task['task_id'], num_results=1, filter='is_error:1')
            event = events.data[0]
            # Print the error event to stderr and Parsl file log if it was not yet printed
            if event['time'] != last_event_time:
                last_event_time = event['time']
                logger.warn('Non-critical Globus Transfer error event for globus://{}{}: "{}" at {}. Retrying...'.format(<commit_msg>Prevent large globus transfers to fail after 60 seconds  (#1323)<commit_after>import logging
import json
import globus_sdk
import os
import parsl
import typeguard

from functools import partial
from typing import Optional
from parsl.app.app import python_app
from parsl.utils import RepresentationMixin
from parsl.data_provider.staging import Staging

logger = logging.getLogger(__name__)

# globus staging must be run explicitly in the same process/interpreter
# as the DFK. it relies on persisting globus state between actions in that
# process.

"""
'Parsl Application' OAuth2 client registered with Globus Auth
by <EMAIL>
"""
CLIENT_ID = '8b8060fd-610e-4a74-885e-1051c71ad473'
REDIRECT_URI = 'https://auth.globus.org/v2/web/auth-code'
SCOPES = ('openid '
          'urn:globus:auth:scope:transfer.api.globus.org:all')


get_input = getattr(__builtins__, 'raw_input', input)


def _get_globus_provider(dfk, executor_label):
    if executor_label is None:
        raise ValueError("executor_label is mandatory")
    executor = dfk.executors[executor_label]
    if not hasattr(executor, "storage_access"):
        raise ValueError("specified executor does not have storage_access attribute")
    for provider in executor.storage_access:
        if isinstance(provider, GlobusStaging):
            return provider

    raise Exception('No suitable Globus endpoint defined for executor {}'.format(executor_label))


def get_globus():
    Globus.init()
    return Globus()


class Globus(object):
    """
    All communication with the Globus Auth and Globus Transfer services is enclosed
    in the Globus class. In particular, the Globus class is reponsible for:
     - managing an OAuth2 authorizer - getting access and refresh tokens,
       refreshing an access token, storing to and retrieving tokens from
       .globus.json file,
     - submitting file transfers,
     - monitoring transfers.
    """

    authorizer = None

    @classmethod
    def init(cls):
        token_path = os.path.join(os.path.expanduser('~'), '.parsl')
        if not os.path.isdir(token_path):
            os.mkdir(token_path)
        cls.TOKEN_FILE = os.path.join(token_path, '.globus.json')

        if cls.authorizer:
            return
        cls.authorizer = cls._get_native_app_authorizer(CLIENT_ID)

    @classmethod
    def get_authorizer(cls):
        return cls.authorizer

    @classmethod
    def transfer_file(cls, src_ep, dst_ep, src_path, dst_path):
        tc = globus_sdk.TransferClient(authorizer=cls.authorizer)
        td = globus_sdk.TransferData(tc, src_ep, dst_ep)
        td.add_item(src_path, dst_path)
        try:
            task = tc.submit_transfer(td)
        except Exception as e:
            raise Exception('Globus transfer from {}{} to {}{} failed due to error: {}'.format(
                src_ep, src_path, dst_ep, dst_path, e))

        last_event_time = None
        """
        A Globus transfer job (task) can be in one of the three states: ACTIVE, SUCCEEDED, FAILED.
        Parsl every 15 seconds polls a status of the transfer job (task) from the Globus Transfer service,
        with 60 second timeout limit. If the task is ACTIVE after time runs out 'task_wait' returns False,
        and True otherwise.
        """
        while not tc.task_wait(task['task_id'], 60, 15):
            task = tc.get_task(task['task_id'])
            # Get the last error Globus event
            events = tc.task_event_list(task['task_id'], num_results=1, filter='is_error:1')
            try:
                event = next(events)
            # No error reported,  the transfer is still running
            except StopIteration:
                continue
            # Print the error event to stderr and Parsl file log if it was not yet printed
            if event['time'] != last_event_time:
                last_event_time = event['time']
                logger.warn('Non-critical Globus Transfer error event for globus://{}{}: "{}" at {}. Retrying...'.format(
{'fraction_non_alphanumeric': 0.08811026664949118, 'fraction_numerical': 0.010176478165657607, 'mean_word_length': 2.9451219512195124, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5190743', 'n_tokens_mistral': 2371, 'n_tokens_neox': 2296, 'n_words': 710}