<commit_before># Copyright (c) 2013 <NAME>
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
# the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.
import abc

import time

import luigi
import luigi.configuration

from mortar.api.v2 import API
from mortar.api.v2 import clusters
from mortar.api.v2 import jobs

import logging
logger = logging.getLogger('luigi-interface')

NUM_MAP_SLOTS_PER_MACHINE = 8
NUM_REDUCE_SLOTS_PER_MACHINE = 3

class MortarProjectTask(luigi.Task):
    
    # default to a cluster of size 2
    cluster_size = luigi.IntParameter(default=2)
    
    # default to a single-job cluster
    cluster_type = luigi.Parameter(default=clusters.CLUSTER_TYPE_SINGLE_JOB)
    
    # run on master by default
    git_ref = luigi.Parameter(default='master')
    
    # Whether to notify on completion of a job
    notify_on_job_finish = luigi.BooleanParameter(default=False)
    
    # interval (in seconds) to poll for job status
    job_polling_interval = luigi.IntParameter(default=5)
        
    @abc.abstractmethod
    def project(self):
        """
        Name of the mortar project to run.
        """
        raise RuntimeError("Must implement project!")

    @abc.abstractmethod
    def script(self):
        """
        Name of the script to run.
        """
        raise RuntimeError("Must implement script!")

    @abc.abstractmethod
    def is_control_script(self):
        """
        Whether this job is a control script.
        """
        raise RuntimeError("Must implement is_control_script!")

    def parameters(self):
        """
        Parameters for this Mortar job.
        """
        return {}

    def run(self):
        """
        Run the mortar job.
        """
        # TODO: parameterize new cluster vs existing cluster
        api = self._get_api()
        job_id = self._run_job(api)
        
        job = self._poll_job_completion(api, job_id)
        final_job_status_code = job.get('status_code')
        if final_job_status_code != jobs.STATUS_SUCCESS:
            raise Exception('Mortar job_id [%s] failed with status_code: [%s], error details: %s' % (job_id, final_job_status_code, job.get('error')))
        else:
            logger.info('Mortar job_id [%s] completed successfully' % job_id)

    def _get_api(self):
        return API(luigi.configuration.get_config().get('mortar', 'email'),
                   luigi.configuration.get_config().get('mortar', 'api_key'))

    def _run_job(self, api):
        job_id = jobs.post_job_new_cluster(api, self.project(), self.script(), self.cluster_size, 
            cluster_type=self.cluster_type, git_ref=self.git_ref, parameters=self.parameters(),
            notify_on_job_finish=self.notify_on_job_finish, is_control_script=self.is_control_script())
        logger.info('Submitted new job to mortar with job_id [%s]' % job_id)
        return job_id<commit_msg>Default to using the largest idle existing cluster that's larger than requested cluster_size<commit_after># Copyright (c) 2013 Mortar Data
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not
# use this file except in compliance with the License. You may obtain a copy of
# the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations under
# the License.
import abc

import time

import luigi
import luigi.configuration

from mortar.api.v2 import API
from mortar.api.v2 import clusters
from mortar.api.v2 import jobs

import logging
logger = logging.getLogger('luigi-interface')

NUM_MAP_SLOTS_PER_MACHINE = 8
NUM_REDUCE_SLOTS_PER_MACHINE = 3

class MortarProjectTask(luigi.Task):
    
    # default to a cluster of size 2
    cluster_size = luigi.IntParameter(default=2)
    
    # whether to run this job on it's own cluster
    # or to use a multi-job cluster
    # if a large enough cluster is running, it will be used,
    # otherwise, a new multi-use cluster will be started
    run_on_single_use_cluster = luigi.BooleanParameter(False)
    
    # run on master by default
    git_ref = luigi.Parameter(default='master')
    
    # Whether to notify on completion of a job
    notify_on_job_finish = luigi.BooleanParameter(default=False)
    
    # interval (in seconds) to poll for job status
    job_polling_interval = luigi.IntParameter(default=5)
        
    @abc.abstractmethod
    def project(self):
        """
        Name of the mortar project to run.
        """
        raise RuntimeError("Must implement project!")

    @abc.abstractmethod
    def script(self):
        """
        Name of the script to run.
        """
        raise RuntimeError("Must implement script!")

    @abc.abstractmethod
    def is_control_script(self):
        """
        Whether this job is a control script.
        """
        raise RuntimeError("Must implement is_control_script!")

    def parameters(self):
        """
        Parameters for this Mortar job.
        """
        return {}

    def run(self):
        """
        Run the mortar job.
        """
        api = self._get_api()
        job_id = self._run_job(api)
        
        job = self._poll_job_completion(api, job_id)
        final_job_status_code = job.get('status_code')
        if final_job_status_code != jobs.STATUS_SUCCESS:
            raise Exception('Mortar job_id [%s] failed with status_code: [%s], error details: %s' % (job_id, final_job_status_code, job.get('error')))
        else:
            logger.info('Mortar job_id [%s] completed successfully' % job_id)

    def _get_api(self):
        return API(luigi.configuration.get_config().get('mortar', 'email'),
                   luigi.configuration.get_config().get('mortar', 'api_key'))

    def _run_job(self, api):
        cluster_type = clusters.CLUSTER_TYPE_SINGLE_JOB if self.run_on_single_use_cluster \
            else clusters.CLUSTER_TYPE_PERSISTENT
        cluster_id = None
        if not self.run_on_single_use_cluster:
            # search for a suitable cluster
            idle_clusters = self._get_idle_clusters(api, min_size=self.cluster_size)
            if idle_clusters:
                # grab the idle largest cluster that's big enough to use
                largest_cluster = sorted(idle_clusters, key=lambda c: int(c['size']), reverse=True)[0]
                logger.info('Using largest running idle cluster with cluster_id [%s], size [%s]' % \
                    (largest_cluster['cluster_id'], largest_cluster['size']))
                cluster_id = largest_cluster['cluster_id']
        
        if cluster_id:
            job_id = jobs.post_job_existing_cluster(api, self.project(), self.script(), cluster_id,
                git_ref=self.git_ref, parameters=self.parameters(),
                notify_on_job_finish=self.notify_on_job_finish, is_control_script=self.is_control_script())
        else:
            job_id = jobs.post_job_new_cluster(api, self.project(), self.script(), self.cluster_size, 
                cluster_type=cluster_type, git_ref=self.git_ref, parameters=self.parameters(),
                notify_on_job_finish=self.notify_on_job_finish, is_control_script=self.is_control_script())
        logger.info('Submitted new job to mortar with job_id [%s]' % job_id)
        return job_id
        
    def _get_idle_clusters(self, api, min_size=0):
        return [cluster for cluster in clusters.get_clusters(api)['clusters'] \
            if (cluster.get('status_code') == clusters.CLUSTER_STATUS_RUNNING) and \
               (cluster.get('cluster_type_code') != clusters.CLUSTER_TYPE_SINGLE_JOB) and \
               (len(cluster.get('running_jobs')) == 0) and \
               (int(cluster.get('size')) >= min_size)]
{'fraction_non_alphanumeric': 0.09294794914847686, 'fraction_numerical': 0.004197649316382826, 'mean_word_length': 2.8840242198416397, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '7456427', 'n_tokens_mistral': 2469, 'n_tokens_neox': 2408, 'n_words': 683}