<commit_before>"""
Task submission directly to condor.
"""
import sys
import os
import logging
import getpass
from datetime import datetime,timedelta
import subprocess
import asyncio
from functools import partial
import gzip

import tornado.gen
from tornado.concurrent import run_on_executor
import requests.exceptions

import iceprod
from iceprod.core import dataclasses
from iceprod.core import constants
from iceprod.core import functions
from iceprod.core.resources import sanitized_requirements
from iceprod.core.exe_json import ServerComms
from iceprod.server import grid
from iceprod.server.globus import SiteGlobusProxy
from iceprod.server.plugins.condor import condor_os_reqs

logger = logging.getLogger('plugin-condor_direct')

async def check_call(*args, **kwargs):
    logger.info('subprocess_check_call: %r', args)
    p = await asyncio.create_subprocess_exec(*args, **kwargs)
    if p.returncode:
        raise Exception(f'command failed, return code {p.returncode}')
    return p

async def check_call_clean_env(*args, **kwargs):
    logger.info('subprocess_check_call: %r', args)
    env = os.environ.copy()
    del env['LD_LIBRARY_PATH']
    kwargs['env'] = env
    p = await asyncio.create_subprocess_exec(*args, **kwargs)
    if p.returncode:
        raise Exception(f'command failed, return code {p.returncode}')
    return p

async def check_output_clean_env(*args, **kwargs):
    logger.info('subprocess_check_output: %r', args)
    kwargs['stdout'] = subprocess.PIPE
    kwargs['stderr'] = subprocess.PIPE
    env = os.environ.copy()
    del env['LD_LIBRARY_PATH']
    kwargs['env'] = env
    p = await asyncio.create_subprocess_exec(*args, **kwargs)
    out,err = await p.communicate()
    out = out+b'\n'+err
    if p.returncode:
        raise Exception(f'command failed, return code {p.returncode}')
    return out.decode('utf-8')

class MyServerComms(ServerComms):
    def __init__(self, rest_client):
        self.rest = rest_client

class TaskInfo(dict):
    def __init__(self, **kwargs):
        self['dataset_id'] = None
        self['job_id'] = None
        self['task_id'] = None
        self['config'] = None
        self['pilot'] = None
        self['submit_dir'] = None
        super(TaskInfo, self).__init__(**kwargs)

def run_async(func,*args,**kwargs):
    async def myfunc():
        return_obj = {'args':args,'kwargs':kwargs}
        try:
            return_obj['return'] = await func(*args,**kwargs)
        except Exception as e:
            logger.info('error running async', exc_info=True)
            return_obj['exception'] = e
        return return_obj
    return asyncio.ensure_future(myfunc())


class condor_direct(grid.BaseGrid):

    ### Plugin Overrides ###

    def __init__(self, *args, **kwargs):
        super(condor_direct, self).__init__(*args, **kwargs)
        self.x509proxy = SiteGlobusProxy()

        # queue requirements
        self.site = 'CondorDirect'
        if 'site' in self.queue_cfg:
            self.site = self.queue_cfg['site']
        self.resources = {'site': self.site}
        if 'gpu' in self.site.lower():
            self.resources['gpu'] = 1
        self.queue_params = {
            'requirements.site': self.site,
        }
        logger.info('resources: %r', self.resources)
        logger.info('queue params: %r', self.queue_params)

        self.grid_remove_once = set()

    async def upload_logfiles(self, task_id, dataset_id, submit_dir='', reason=''):
        """upload logfiles"""
        data = {'name': 'stdlog', 'task_id': task_id, 'dataset_id': dataset_id}

        def read_filename(filename):<commit_msg>allow site exclusivity to be configurable<commit_after>"""
Task submission directly to condor.
"""
import sys
import os
import logging
import getpass
from datetime import datetime,timedelta
import subprocess
import asyncio
from functools import partial
import gzip

import tornado.gen
from tornado.concurrent import run_on_executor
import requests.exceptions

import iceprod
from iceprod.core import dataclasses
from iceprod.core import constants
from iceprod.core import functions
from iceprod.core.resources import sanitized_requirements
from iceprod.core.exe_json import ServerComms
from iceprod.server import grid
from iceprod.server.globus import SiteGlobusProxy
from iceprod.server.plugins.condor import condor_os_reqs

logger = logging.getLogger('plugin-condor_direct')

async def check_call(*args, **kwargs):
    logger.info('subprocess_check_call: %r', args)
    p = await asyncio.create_subprocess_exec(*args, **kwargs)
    if p.returncode:
        raise Exception(f'command failed, return code {p.returncode}')
    return p

async def check_call_clean_env(*args, **kwargs):
    logger.info('subprocess_check_call: %r', args)
    env = os.environ.copy()
    del env['LD_LIBRARY_PATH']
    kwargs['env'] = env
    p = await asyncio.create_subprocess_exec(*args, **kwargs)
    if p.returncode:
        raise Exception(f'command failed, return code {p.returncode}')
    return p

async def check_output_clean_env(*args, **kwargs):
    logger.info('subprocess_check_output: %r', args)
    kwargs['stdout'] = subprocess.PIPE
    kwargs['stderr'] = subprocess.PIPE
    env = os.environ.copy()
    del env['LD_LIBRARY_PATH']
    kwargs['env'] = env
    p = await asyncio.create_subprocess_exec(*args, **kwargs)
    out,err = await p.communicate()
    out = out+b'\n'+err
    if p.returncode:
        raise Exception(f'command failed, return code {p.returncode}')
    return out.decode('utf-8')

class MyServerComms(ServerComms):
    def __init__(self, rest_client):
        self.rest = rest_client

class TaskInfo(dict):
    def __init__(self, **kwargs):
        self['dataset_id'] = None
        self['job_id'] = None
        self['task_id'] = None
        self['config'] = None
        self['pilot'] = None
        self['submit_dir'] = None
        super(TaskInfo, self).__init__(**kwargs)

def run_async(func,*args,**kwargs):
    async def myfunc():
        return_obj = {'args':args,'kwargs':kwargs}
        try:
            return_obj['return'] = await func(*args,**kwargs)
        except Exception as e:
            logger.info('error running async', exc_info=True)
            return_obj['exception'] = e
        return return_obj
    return asyncio.ensure_future(myfunc())


class condor_direct(grid.BaseGrid):

    ### Plugin Overrides ###

    def __init__(self, *args, **kwargs):
        super(condor_direct, self).__init__(*args, **kwargs)
        self.x509proxy = SiteGlobusProxy()

        # queue requirements
        self.site = 'CondorDirect'
        if 'site' in self.queue_cfg:
            self.site = self.queue_cfg['site']
        self.resources = {'site': self.site}
        if 'gpu' in self.site.lower():
            self.resources['gpu'] = 1
        self.queue_params = {}
        if 'exclusive' in self.queue_cfg and self.queue_cfg['exclusive']:
            self.queue_params['requirements.site'] = self.site
        logger.info('resources: %r', self.resources)
        logger.info('queue params: %r', self.queue_params)

        self.grid_remove_once = set()

    async def upload_logfiles(self, task_id, dataset_id, submit_dir='', reason=''):
        """upload logfiles"""
        data = {'name': 'stdlog', 'task_id': task_id, 'dataset_id': dataset_id}

        def read_filename(filename):
{'fraction_non_alphanumeric': 0.1204967248908297, 'fraction_numerical': 0.0013646288209606986, 'mean_word_length': 3.4607425441265978, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, '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': '353879', 'n_tokens_mistral': 2252, 'n_tokens_neox': 2217, 'n_words': 548}