<commit_before>"""
Plugin instance app manager module that provides functionality to run and check the
execution status of a plugin instance's app (ChRIS / pfcon interface).

NOTE:

    This module is executed as part of an asynchronous celery worker.
    For instance, to debug 'check_plugin_instance_app_exec_status' method synchronously
    with pudb.set_trace() you need to:

    1. Once CUBE is running, and assuming some plugininstance has been POSTed, start a
    python shell on the manage.py code (note <IMAGE> below is the chris:dev container):

    docker exec -ti <IMAGE> python manage.py shell

    You should now be in a python shell.

    2. To simulate operations on a given plugin with id <id>,
    instantiate the relevant objects (for ex, for id=1):

    from plugininstances.models import PluginInstance
    from plugininstances.services import manager

    plg_inst = PluginInstance.objects.get(id=1)
    plg_inst_manager = manager.PluginInstanceManager(plg_inst)

    3. And finally, call the method:

    plg_inst_manager.check_plugin_instance_app_exec_status()

    Any pudb.set_trace() calls in this method will now be handled by the pudb debugger.

    4. Finally, after each change to this method, reload this module:

    import importlib
    importlib.reload(manager)

    and also re-instantiate the service:

    plg_inst_manager = manager.PluginInstanceManager(plg_inst)
"""

import logging
import os
import re
import io
import time
import json
import zipfile
from typing import List, Optional
from pfconclient import client as pfcon
from pfconclient.exceptions import (PfconRequestException,
                                    PfconRequestInvalidTokenException)

from django.utils import timezone
from django.conf import settings
from django.db.utils import IntegrityError

from core.swiftmanager import SwiftManager, ClientException
from core.utils import json_zip2str
from core.models import ChrisInstance
from plugininstances.models import PluginInstance, PluginInstanceFile, PluginInstanceLock

if settings.DEBUG:
    import pdb, pudb, rpudb
    from celery.contrib import rdb


logger = logging.getLogger(__name__)


class PluginInstanceManager(object):

    def __init__(self, plugin_instance):

        self.c_plugin_inst = plugin_instance

        self.l_plugin_inst_param_instances = self.c_plugin_inst.get_parameter_instances()

        self.str_job_id = ChrisInstance.load().job_id_prefix + str(plugin_instance.id)

        cr = self.c_plugin_inst.compute_resource
        self.pfcon_client = pfcon.Client(cr.compute_url, cr.compute_auth_token)

        self.swift_manager = SwiftManager(settings.SWIFT_CONTAINER_NAME,
                                          settings.SWIFT_CONNECTION_PARAMS)

    def _refresh_compute_resource_auth_token(self):
        """
        Get a new auth token from a remote pfcon service and update the DB.
        """
        cr = self.c_plugin_inst.compute_resource
        token = pfcon.Client.get_auth_token(cr.compute_auth_url, cr.compute_user,
                                            cr.compute_password)
        self.pfcon_client.set_auth_token(token)
        cr.compute_auth_token = token
        cr.save()

    def run_plugin_instance_app(self):
        """
        Run the plugin instance's app via a call to a remote pfcon service.
        """
        if self.c_plugin_inst.status == 'cancelled':
            return

        plugin = self.c_plugin_inst.plugin
        plugin_type = plugin.meta.type
        inputdirs = []
        try:
            if plugin_type == 'ds':
                inputdirs.append(self.get_previous_output_path())
            else:
                inputdirs.append(self.manage_plugin_instance_app_empty_inputdir())
        except Exception:
            self.c_plugin_inst.status = 'cancelled'  # giving up
            self.save_plugin_instance_final_status()
            return

        d_unextpath_params, d_path_params = self.get_plugin_instance_path_parameters()
        for path_param_value in [param_value for param_value in d_path_params.values()]:
            # the value of each parameter of type 'path' is a string
            # representing a comma-separated list of paths in obj storage
            inputdirs = inputdirs + path_param_value.split(',')

        # create data file to transmit
        try:
            zip_file = self.create_zip_file(inputdirs)
        except Exception:
            self.c_plugin_inst.status = 'cancelled'  # giving up
            self.save_plugin_instance_final_status()
            return

        # create job description dictionary
        job_descriptors = {
            'entrypoint': self._assemble_exec(plugin.selfpath, plugin.selfexec, plugin.execshell),
            'args': self.get_plugin_instance_app_cmd_args(),
            'args_path_flags': [*d_unextpath_params.keys(), *d_path_params.keys()],
            'auid': self.c_plugin_inst.owner.username,
            'number_of_workers': self.c_plugin_inst.number_of_workers,
            'cpu_limit': self.c_plugin_inst.cpu_limit,
            'memory_limit': self.c_plugin_inst.memory_limit,
            'gpu_limit': self.c_plugin_inst.gpu_limit,
            'image': plugin.dock_image,
            'type': plugin_type
        }
        pfcon_url = self.pfcon_client.url
        job_id = self.str_job_id
        logger.info(f'Submitting job {job_id} to pfcon url -->{pfcon_url}<--, '
                    f'description: {json.dumps(job_descriptors, indent=4)}')
        try:
            d_resp = self._submit_job(job_id, job_descriptors, zip_file.getvalue())
        except PfconRequestException as e:
            logger.error(f'[CODE01,{job_id}]: Error submitting job to pfcon url '
                         f'-->{pfcon_url}<--, detail: {str(e)}')
            self.c_plugin_inst.error_code = 'CODE01'
            self.c_plugin_inst.status = 'cancelled'  # giving up
            self.save_plugin_instance_final_status()
        else:
            logger.info(f'Successfully submitted job {job_id} to pfcon url '
                        f'-->{pfcon_url}<--, response: {json.dumps(d_resp, indent=4)}')
            # update the job status and summary
            self.c_plugin_inst.status = 'started'
            self.c_plugin_inst.summary = self.get_job_status_summary()  # initial status
            self.c_plugin_inst.raw = json_zip2str(d_resp)
            self.c_plugin_inst.save()

    @staticmethod
    def _assemble_exec(selfpath: Optional[str], selfexec: str, execshell: Optional[str]) -> List[str]:
        entrypoint_path = selfexec
        if selfpath:
            entrypoint_path = os.path.join(selfpath, selfexec)
        if not execshell:
            return [entrypoint_path]
        return [execshell, entrypoint_path]

    def _submit_job(self, job_id, job_descriptors, dfile, timeout=9000):
        """
        Submit job to a remote pfcon service.
        """
        try:
            d_resp = self.pfcon_client.submit_job(job_id, job_descriptors, dfile, timeout)
        except PfconRequestInvalidTokenException:
            logger.info(f'Auth token has expired while submitting job {job_id} to pfcon '
                        f'url -->{self.pfcon_client.url}<--')
            self._refresh_compute_resource_auth_token()
            d_resp = self.pfcon_client.submit_job(job_id, job_descriptors, dfile, timeout)
        return d_resp

    def check_plugin_instance_app_exec_status(self):
        """
        Check a plugin instance's app execution status. If the associated job's
        execution time exceeds the maximum set for the remote compute environment then
        the job is cancelled. Otherwise the job's execution status is fetched from the
        remote and if finished without error then the job's zip file is downloaded and
        unpacked and the output files registered with the DB. Finally a delete request
        is made to remove the job from the remote environment.
        """
        if self.c_plugin_inst.status == 'started':
            job_id = self.str_job_id

            delta_exec_time = timezone.now() - self.c_plugin_inst.start_date
            delta_seconds = delta_exec_time.total_seconds()
            max_exec_seconds = self.c_plugin_inst.compute_resource.max_job_exec_seconds
            if delta_seconds > max_exec_seconds:
                logger.error(f'[CODE13,{job_id}]: Error, job exceeded maximum execution '
                             f'time ({max_exec_seconds} seconds)')<commit_msg>Workaround for connection aborted<commit_after>"""
Plugin instance app manager module that provides functionality to run and check the
execution status of a plugin instance's app (ChRIS / pfcon interface).

NOTE:

    This module is executed as part of an asynchronous celery worker.
    For instance, to debug 'check_plugin_instance_app_exec_status' method synchronously
    with pudb.set_trace() you need to:

    1. Once CUBE is running, and assuming some plugininstance has been POSTed, start a
    python shell on the manage.py code (note <IMAGE> below is the chris:dev container):

    docker exec -ti <IMAGE> python manage.py shell

    You should now be in a python shell.

    2. To simulate operations on a given plugin with id <id>,
    instantiate the relevant objects (for ex, for id=1):

    from plugininstances.models import PluginInstance
    from plugininstances.services import manager

    plg_inst = PluginInstance.objects.get(id=1)
    plg_inst_manager = manager.PluginInstanceManager(plg_inst)

    3. And finally, call the method:

    plg_inst_manager.check_plugin_instance_app_exec_status()

    Any pudb.set_trace() calls in this method will now be handled by the pudb debugger.

    4. Finally, after each change to this method, reload this module:

    import importlib
    importlib.reload(manager)

    and also re-instantiate the service:

    plg_inst_manager = manager.PluginInstanceManager(plg_inst)
"""

import logging
import os
import re
import io
import time
import json
import zipfile
from typing import List, Optional
from pfconclient import client as pfcon
from pfconclient.exceptions import (PfconRequestException,
                                    PfconRequestInvalidTokenException)

from django.utils import timezone
from django.conf import settings
from django.db.utils import IntegrityError

from core.swiftmanager import SwiftManager, ClientException
from core.utils import json_zip2str
from core.models import ChrisInstance
from plugininstances.models import PluginInstance, PluginInstanceFile, PluginInstanceLock

if settings.DEBUG:
    import pdb, pudb, rpudb
    from celery.contrib import rdb


logger = logging.getLogger(__name__)


class PluginInstanceManager(object):

    def __init__(self, plugin_instance):

        self.c_plugin_inst = plugin_instance

        self.l_plugin_inst_param_instances = self.c_plugin_inst.get_parameter_instances()

        self.str_job_id = ChrisInstance.load().job_id_prefix + str(plugin_instance.id)

        cr = self.c_plugin_inst.compute_resource
        self.pfcon_client = pfcon.Client(cr.compute_url, cr.compute_auth_token)

        self.swift_manager = SwiftManager(settings.SWIFT_CONTAINER_NAME,
                                          settings.SWIFT_CONNECTION_PARAMS)

    def _refresh_compute_resource_auth_token(self):
        """
        Get a new auth token from a remote pfcon service and update the DB.
        """
        cr = self.c_plugin_inst.compute_resource
        token = pfcon.Client.get_auth_token(cr.compute_auth_url, cr.compute_user,
                                            cr.compute_password)
        self.pfcon_client.set_auth_token(token)
        cr.compute_auth_token = token
        cr.save()

    def run_plugin_instance_app(self):
        """
        Run the plugin instance's app via a call to a remote pfcon service.
        """
        if self.c_plugin_inst.status == 'cancelled':
            return

        plugin = self.c_plugin_inst.plugin
        plugin_type = plugin.meta.type
        inputdirs = []
        try:
            if plugin_type == 'ds':
                inputdirs.append(self.get_previous_output_path())
            else:
                inputdirs.append(self.manage_plugin_instance_app_empty_inputdir())
        except Exception:
            self.c_plugin_inst.status = 'cancelled'  # giving up
            self.save_plugin_instance_final_status()
            return

        d_unextpath_params, d_path_params = self.get_plugin_instance_path_parameters()
        for path_param_value in [param_value for param_value in d_path_params.values()]:
            # the value of each parameter of type 'path' is a string
            # representing a comma-separated list of paths in obj storage
            inputdirs = inputdirs + path_param_value.split(',')

        # create data file to transmit
        try:
            zip_file = self.create_zip_file(inputdirs)
        except Exception:
            self.c_plugin_inst.status = 'cancelled'  # giving up
            self.save_plugin_instance_final_status()
            return

        # create job description dictionary
        job_descriptors = {
            'entrypoint': self._assemble_exec(plugin.selfpath, plugin.selfexec, plugin.execshell),
            'args': self.get_plugin_instance_app_cmd_args(),
            'args_path_flags': [*d_unextpath_params.keys(), *d_path_params.keys()],
            'auid': self.c_plugin_inst.owner.username,
            'number_of_workers': self.c_plugin_inst.number_of_workers,
            'cpu_limit': self.c_plugin_inst.cpu_limit,
            'memory_limit': self.c_plugin_inst.memory_limit,
            'gpu_limit': self.c_plugin_inst.gpu_limit,
            'image': plugin.dock_image,
            'type': plugin_type
        }
        pfcon_url = self.pfcon_client.url
        job_id = self.str_job_id
        logger.info(f'Submitting job {job_id} to pfcon url -->{pfcon_url}<--, '
                    f'description: {json.dumps(job_descriptors, indent=4)}')
        try:
            d_resp = self._submit_job(job_id, job_descriptors, zip_file.getvalue())
        except PfconRequestException as e:
            logger.error(f'[CODE01,{job_id}]: Error submitting job to pfcon url '
                         f'-->{pfcon_url}<--, detail: {str(e)}')
            self.c_plugin_inst.error_code = 'CODE01'
            self.c_plugin_inst.status = 'cancelled'  # giving up
            self.save_plugin_instance_final_status()
        else:
            logger.info(f'Successfully submitted job {job_id} to pfcon url '
                        f'-->{pfcon_url}<--, response: {json.dumps(d_resp, indent=4)}')
            # update the job status and summary
            self.c_plugin_inst.status = 'started'
            self.c_plugin_inst.summary = self.get_job_status_summary()  # initial status
            self.c_plugin_inst.raw = json_zip2str(d_resp)
            self.c_plugin_inst.save()

    @staticmethod
    def _assemble_exec(selfpath: Optional[str], selfexec: str, execshell: Optional[str]) -> List[str]:
        entrypoint_path = selfexec
        if selfpath:
            entrypoint_path = os.path.join(selfpath, selfexec)
        if not execshell:
            return [entrypoint_path]
        return [execshell, entrypoint_path]

    def _submit_job(self, job_id, job_descriptors, dfile, timeout=9000):
        """
        Submit job to a remote pfcon service.
        """
        try:
            d_resp = self.pfcon_client.submit_job(job_id, job_descriptors, dfile, timeout)
        except PfconRequestException:
            # FIXME HACK
            # Under some conditions, the requests library will produce a "Connection Aborted"
            # error instead of a 401 response. This happens when pfcon responds eagerly
            # to an invalid token and closes the connection while CUBE is trying to transmit
            # a large zip file.
            # The temporary workaround is to catch a wider range of Exceptions here.
            # Ideally we only want to try again in the event that we know the token is
            # invalid, PfconRequestInvalidTokenException, however PfconRequestInvalidTokenException
            # is not correctly raised in all the situations where it should be.
            #
            # logger.info(f'Auth token has expired while submitting job {job_id} to pfcon '
            logger.exception(f'Error while submitting job {job_id} to pfcon '
                             f'url -->{self.pfcon_client.url}<--')
            logger.info(f'Will retry to submit job {job_id}')
            self._refresh_compute_resource_auth_token()
            d_resp = self.pfcon_client.submit_job(job_id, job_descriptors, dfile, timeout)
        return d_resp

    def check_plugin_instance_app_exec_status(self):
        """
        Check a plugin instance's app execution status. If the associated job's
        execution time exceeds the maximum set for the remote compute environment then
        the job is cancelled. Otherwise the job's execution status is fetched from the
        remote and if finished without error then the job's zip file is downloaded and
        unpacked and the output files registered with the DB. Finally a delete request
        is made to remove the job from the remote environment.
        """
        if self.c_plugin_inst.status == 'started':
            job_id = self.str_job_id

            delta_exec_time = timezone.now() - self.c_plugin_inst.start_date
            delta_seconds = delta_exec_time.total_seconds()
            max_exec_seconds = self.c_plugin_inst.compute_resource.max_job_exec_seconds
            if delta_seconds > max_exec_seconds:
                logger.error(f'[CODE13,{job_id}]: Error, job exceeded maximum execution '
                             f'time ({max_exec_seconds} seconds)')
{'fraction_non_alphanumeric': 0.07320092274798852, 'fraction_numerical': 0.002419400213807461, 'mean_word_length': 2.6221724067658445, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2552547', 'n_tokens_mistral': 5124, 'n_tokens_neox': 4943, 'n_words': 1362}