<commit_before>import nomad
import socket
import time
from nomad import Nomad
from nomad.api.exceptions import URLNotFoundNomadException
from typing import Callable, List, Set
from threading import Thread
from functools import wraps
from retrying import retry
from datetime import datetime, timedelta
from django.conf import settings
from django.utils import timezone
from django.db import transaction
from data_refinery_common.models import (
    DownloaderJob,
    ProcessorJob,
    SurveyJob,
    DownloaderJobOriginalFileAssociation,
    ProcessorJobOriginalFileAssociation,
    ProcessorJobDatasetAssociation,
    SurveyJobKeyValue
)
from data_refinery_common.message_queue import send_job
from data_refinery_common.job_lookup import (
    Downloaders,
    ProcessorPipeline,
    SurveyJobTypes,
    does_processor_job_have_samples,
    is_file_rnaseq,
)
from data_refinery_common.logging import get_and_configure_logger
from data_refinery_common.utils import (
    get_active_volumes,
    get_env_variable,
    get_env_variable_gracefully
)


logger = get_and_configure_logger(__name__)

# Maximum number of retries, so the number of attempts will be one
# greater than this because of the first attempt
MAX_NUM_RETRIES = 2

# This can be overritten by the env var "MAX_TOTAL_JOBS"
DEFAULT_MAX_JOBS = 20000

# The fastest each thread will repeat its checks.
# Could be slower if the thread takes longer than this to check its jobs.
MIN_LOOP_TIME = timedelta(minutes=2)

# The amount of time the main loop will wait in between checking if
# threads are still alive and then heart beating.
THREAD_WAIT_TIME = timedelta(minutes=10)

# How frequently we dispatch Janitor jobs.
JANITOR_DISPATCH_TIME = timedelta(minutes=30)

# How frequently we clean unplaceable jobs out of the Nomad queue.
CLEANUP_QUEUE_TIME = timedelta(minutes=30)


def read_config_list(config_file: str) -> List[str]:
    """
    Reads a file and returns a list with one item per line.
    """
    return_list = []
    with open(config_file) as config_list:
        for line in config_list:
            return_list.append(line.strip())

    return return_list


# These are lists of accessions that are for
# subsets of experiments that we want to prioritize.
PEDIATRIC_ACCESSION_LIST = read_config_list("config/pediatric_accessions.txt")
HGU133PLUS2_ACCESSION_LIST = read_config_list("config/hgu133plus2_accessions.txt")


##
# Utilities
##

def do_forever(min_loop_time: timedelta) -> Callable:
    """Run the wrapped function in a loop forever.

    The function won't be run more often than once per min_loop_time,
    however if it takes longer to run than min_loop_time, then it will
    be run less often than once per min_loop_time.
    """
    def decorator(function: Callable) -> Callable:
        @wraps(function)
        def wrapper(*args, **kwargs):
            while(True):
                start_time = timezone.now()

                try:
                    function(*args, **kwargs)
                except:
                    logger.exception("Exception caught by Foreman while running " + function.__name__)

                loop_time = timezone.now() - start_time
                if loop_time < min_loop_time:
                    remaining_time = min_loop_time - loop_time
                    time.sleep(remaining_time.seconds)

        return wrapper
    return decorator


def handle_repeated_failure(job) -> None:
    """If a job fails too many times, log it and stop retrying."""
    # Not strictly retried but will prevent the job from getting
    # retried any more times.
    job.retried = True

    # success may already be False, but if it was a hung or lost job
    # this will ensure it's marked as failed.
    job.success = False
    job.save()

    # At some point this should become more noisy/attention
    # grabbing. However for the time being just logging should be
    # sufficient because all log messages will be closely monitored
    # during early testing stages.
    logger.warn("%s #%d failed %d times!!!", job.__class__.__name__, job.id, MAX_NUM_RETRIES + 1)


##
# Job Prioritization
##


def prioritize_salmon_jobs(jobs: List) -> List:
    """Prioritizes salmon experiments based on how close to completion they are.

    This is because salmon experiments have a final processing step
    that must be performed on all the samples in the experiment, so if
    9/10 samples in an experiment are processed then they can't
    actually be used until that last sample is processed.
    """
    # The strategy for doing so is to build a mapping between every
    # salmon job in `jobs` to a priority. This priority will be what
    # percentage of the samples in this experiment have been
    # processed. Once we have that mapping we can sort those jobs by
    # that priority and move them to the front of the list.
    prioritized_jobs = []
    for job in jobs:
        try:
            if not does_processor_job_have_samples(job):
                continue

            # Salmon jobs are specifc to one sample.
            sample = job.get_samples().pop()

            # Skip jobs that aren't for Salmon. Handle both ProcessorJobs and DownloaderJobs.
            if type(job) is ProcessorJob and job.pipeline_applied != ProcessorPipeline.SALMON.value:
                continue
            elif type(job) is DownloaderJob:
                is_salmon_sample = False
                for original_file in sample.original_files.all():
                    if is_file_rnaseq(original_file.filename):
                        is_salmon_sample = True

                if not is_salmon_sample:
                    continue

            # Get a set of unique samples that share at least one
            # experiment with the sample this job is for.
            related_samples = set()
            for experiment in sample.experiments.all():
                for related_sample in experiment.samples.all():
                    related_samples.add(related_sample)

            # We cannot simply filter on is_processed because that field
            # doesn't get set until every sample in an experiment is processed.
            # Instead we are looking for one successful processor job.
            processed_samples = 0
            for related_sample in related_samples:
                original_files = related_sample.original_files
                if original_files.count() == 0:
                    logger.error("Salmon sample found without any original files!!!", sample=related_sample)
                elif original_files.first().processor_jobs.filter(success=True).count() >= 1:
                    processed_samples += 1

            experiment_completion_percent = processed_samples / len(related_samples)
            prioritized_jobs.append({"job": job, "priority": experiment_completion_percent})
        except:
            logger.debug("Exception caught while prioritizing salmon jobs!", job=job)


    sorted_job_mappings = sorted(prioritized_jobs, reverse=True, key=lambda k: k["priority"])
    sorted_jobs = [job_mapping["job"] for job_mapping in sorted_job_mappings]

    # Remove all the jobs we're moving to the front of the list
    for job in sorted_jobs:
        jobs.remove(job)

    return sorted_jobs + jobs


def prioritize_zebrafish_jobs(jobs: List) -> List:
    """Moves zebrafish jobs to the beginnging of the input list."""
    zebrafish_jobs = []
    for job in jobs:
        try:
            if not does_processor_job_have_samples(job):
                continue

            # There aren't cross-species jobs, so just checking one sample's organism will be sufficient.
            samples = job.get_samples()

            for sample in samples:
                if sample.organism.name == 'DANIO_RERIO':
                    zebrafish_jobs.append(job)
                    break
        except:
            logger.debug("Exception caught while prioritizing zebrafish jobs!", job=job)

    # Remove all the jobs we're moving to the front of the list
    for job in zebrafish_jobs:
        jobs.remove(job)

    return zebrafish_jobs + jobs


def prioritize_jobs_by_accession(jobs: List, accession_list: List[str]) -> List:
    """Moves jobs whose accessions are in accession_lst to the beginning of the input list."""
    prioritized_jobs = []
    for job in jobs:
        try:
            if not does_processor_job_have_samples(job):
                continue

            # All samples in a job correspond to the same experiment, so just check one sample.
            samples = job.get_samples()

            # Iterate through all the samples' experiments until one is
            # found with an accession in `accession_list`.
            is_prioritized_job = False
            for sample in samples:
                if is_prioritized_job:
                    # We found one! So stop looping
                    break

                for experiment in sample.experiments.all():
                    if experiment.accession_code in accession_list:
                        prioritized_jobs.append(job)
                        is_prioritized_job = True
                        break
        except:
            logger.exception("Exception caught while prioritizing jobs by accession!", job=job)

    # Remove all the jobs we're moving to the front of the list
    for job in prioritized_jobs:
        jobs.remove(job)

    return prioritized_jobs + jobs


##
# Downloaders
##


@retry(stop_max_attempt_number=3)
@transaction.atomic
def requeue_downloader_job(last_job: DownloaderJob) -> None:
    """Queues a new downloader job.

    The new downloader job will have num_retries one greater than
    last_job.num_retries.
    """
    num_retries = last_job.num_retries + 1

    new_job = DownloaderJob(num_retries=num_retries,
                            downloader_task=last_job.downloader_task,
                            accession_code=last_job.accession_code)
    new_job.save()

    for original_file in last_job.original_files.all():
        DownloaderJobOriginalFileAssociation.objects.get_or_create(downloader_job=new_job,
                                                           original_file=original_file)

    logger.debug("Requeuing Downloader Job which had ID %d with a new Downloader Job with ID %d.",
                 last_job.id,
                 new_job.id)
    try:
        if send_job(Downloaders[last_job.downloader_task], job=new_job, is_dispatch=True):
            last_job.retried = True
            last_job.success = False
            last_job.retried_job = new_job
            last_job.save()
        else:
            # Can't communicate with nomad just now, leave the job for a later loop.
            new_job.delete()
    except:
        logger.error("Failed to requeue Downloader Job which had ID %d with a new Downloader Job with ID %d.",
                     last_job.id,
                     new_job.id)
        # Can't communicate with nomad just now, leave the job for a later loop.
        new_job.delete()


def handle_downloader_jobs(jobs: List[DownloaderJob]) -> None:
    """For each job in jobs, either retry it or log it."""

    nomad_host = get_env_variable("NOMAD_HOST")
    nomad_port = get_env_variable("NOMAD_PORT", "4646")
    nomad_client = Nomad(nomad_host, port=int(nomad_port), timeout=30)
    # Maximum number of total jobs running at a time.
    # We do this now rather than import time for testing purposes.
    MAX_TOTAL_JOBS = int(get_env_variable_gracefully("MAX_TOTAL_JOBS", DEFAULT_MAX_JOBS))
    len_all_jobs = len(nomad_client.jobs.get_jobs())
    if len_all_jobs >= MAX_TOTAL_JOBS:
        logger.info("Not requeuing job until we're running fewer jobs.")
        return False

    # We want zebrafish data first, then hgu133plus2, then data
    # related to pediatric cancer, then to finish salmon experiments
    # that are close to completion.
    # Each function moves the jobs it prioritizes to the front of the
    # list, so apply them in backwards order.
    # jobs = prioritize_salmon_jobs(jobs)
    # jobs = prioritize_jobs_by_accession(jobs, PEDIATRIC_ACCESSION_LIST)
    # jobs = prioritize_jobs_by_accession(jobs, HGU133PLUS2_ACCESSION_LIST)
    # jobs = prioritize_zebrafish_jobs(jobs)

    jobs_dispatched = 0
    for count, job in enumerate(jobs):
        if job.num_retries < MAX_NUM_RETRIES:
            requeue_downloader_job(job)
            jobs_dispatched = jobs_dispatched + 1
        else:
            handle_repeated_failure(job)

        if (count % 100) == 0:
            len_all_jobs = len(nomad_client.jobs.get_jobs())

        if (jobs_dispatched + len_all_jobs) >= MAX_TOTAL_JOBS:
            logger.info("We hit the maximum total jobs ceiling, so we're not handling any more downloader jobs now.")
            return False

    return True

@do_forever(MIN_LOOP_TIME)
def retry_failed_downloader_jobs() -> None:
    """Handle downloader jobs that were marked as a failure."""
    failed_jobs = DownloaderJob.objects.filter(
        success=False,
        retried=False
    ).prefetch_related(
        "original_files__samples"
    )
    failed_jobs_list = [job for job in failed_jobs]

    if failed_jobs_list:
        logger.info(
            "Handling failed (explicitly-marked-as-failure) jobs!",
            jobs_count=len(failed_jobs_list)
        )
        handle_downloader_jobs(failed_jobs_list)


@do_forever(MIN_LOOP_TIME)
def retry_hung_downloader_jobs() -> None:
    """Retry downloader jobs that were started but never finished."""
    potentially_hung_jobs = DownloaderJob.objects.filter(
        success=None,
        retried=False,
        end_time=None,
        start_time__isnull=False,
        no_retry=False
    ).prefetch_related(
        "original_files__samples"
    )

    nomad_host = get_env_variable("NOMAD_HOST")
    nomad_port = get_env_variable("NOMAD_PORT", "4646")
    nomad_client = Nomad(nomad_host, port=int(nomad_port), timeout=30)
    hung_jobs = []
    for job in potentially_hung_jobs:
        try:
            job_status = nomad_client.job.get_job(job.nomad_job_id)["Status"]
            if job_status != "running":
                # Make sure it didn't finish since our original query.
                job.refresh_from_db()
                if job.end_time is None:
                    hung_jobs.append(job)
        except URLNotFoundNomadException:
            hung_jobs.append(job)
        except Exception:
            logger.exception("Couldn't query Nomad about Processor Job.", processor_job=job.id)

    handle_downloader_jobs(hung_jobs)


@do_forever(MIN_LOOP_TIME)
def retry_lost_downloader_jobs() -> None:
    """Retry downloader jobs that went too long without being started.

    Idea: at some point this function could integrate with the spot
    instances to determine if jobs are hanging due to a lack of
    instances. A naive time-based implementation like this could end
    up retrying every single queued job if there were a long period
    during which the price of spot instance is higher than our bid
    price.
    """
    potentially_lost_jobs = DownloaderJob.objects.filter(
        success=None,
        retried=False,
        start_time=None,
        end_time=None,
        no_retry=False
    ).prefetch_related(
        "original_files__samples"
    )

    nomad_host = get_env_variable("NOMAD_HOST")
    nomad_port = get_env_variable("NOMAD_PORT", "4646")
    nomad_client = Nomad(nomad_host, port=int(nomad_port), timeout=30)
    lost_jobs = []
    for job in potentially_lost_jobs:
        try:
            if job.nomad_job_id:
                job_status = nomad_client.job.get_job(job.nomad_job_id)["Status"]
                # If the job is still pending, then it makes sense that it
                # hasn't started and if it's running then it may not have
                # been able to mark the job record as started yet.
                if job_status != "pending" and job_status != "running":
                    logger.debug(("Determined that a downloader job needs to be requeued because its"
                                  " Nomad Job's status is: %s."),
                                 job_status,
                                 job_id=job.id
                    )
                    lost_jobs.append(job)
            else:
                # If there is no nomad_job_id field set, we could be
                # in the small window where the job was created but
                # hasn't yet gotten a chance to be queued.
                # If this job really should be restarted we'll get it in the next loop.
                if timezone.now() - job.created_at > MIN_LOOP_TIME:
                    lost_jobs.append(job)
        except socket.timeout:
            logger.info("Timeout connecting to Nomad - is Nomad down?", job_id=job.id)
        except URLNotFoundNomadException:
            logger.debug(("Determined that a downloader job needs to be requeued because "
                              "querying for its Nomad job failed: "),
                             job_id=job.id
            )
            lost_jobs.append(job)
        except nomad.api.exceptions.BaseNomadException:
            logger.info("Problem connecting to Nomad - is Nomad down?", job_id=job.id)
        except Exception:
            logger.exception("Couldn't query Nomad about Processor Job.", processor_job=job.id)

    handle_downloader_jobs(lost_jobs)


##
# Processors
##


@retry(stop_max_attempt_number=3)
@transaction.atomic
def requeue_processor_job(last_job: ProcessorJob) -> None:
    """Queues a new processor job.

    The new processor job will have num_retries one greater than
    last_job.num_retries.
    """
    num_retries = last_job.num_retries + 1

    # The Salmon pipeline is quite RAM-sensitive.
    # Try it again with an increased RAM amount, if possible.
    new_ram_amount = last_job.ram_amount
    if last_job.pipeline_applied == "SALMON":
        if new_ram_amount == 4096:
            new_ram_amount = 8192
        elif new_ram_amount == 8192:
            new_ram_amount = 12288
        elif new_ram_amount == 12288:
            new_ram_amount = 16384

    new_job = ProcessorJob(num_retries=num_retries,
                           pipeline_applied=last_job.pipeline_applied,
                           ram_amount=new_ram_amount,
                           volume_index=last_job.volume_index)
    new_job.save()

    for original_file in last_job.original_files.all():
        ProcessorJobOriginalFileAssociation.objects.get_or_create(processor_job=new_job,
                                                          original_file=original_file)

    for dataset in last_job.datasets.all():
        ProcessorJobDatasetAssociation.objects.get_or_create(processor_job=new_job,
                                                     dataset=dataset)

    try:
        logger.debug("Requeuing Processor Job which had ID %d with a new Processor Job with ID %d.",
                     last_job.id,
                     new_job.id)
        if send_job(ProcessorPipeline[last_job.pipeline_applied], job=new_job, is_dispatch=True):
            last_job.retried = True
            last_job.success = False
            last_job.retried_job = new_job<commit_msg>Increase RAM amounts when retrying failed AFFY jobs<commit_after>import nomad
import socket
import time
from nomad import Nomad
from nomad.api.exceptions import URLNotFoundNomadException
from typing import Callable, List, Set
from threading import Thread
from functools import wraps
from retrying import retry
from datetime import datetime, timedelta
from django.conf import settings
from django.utils import timezone
from django.db import transaction
from data_refinery_common.models import (
    DownloaderJob,
    ProcessorJob,
    SurveyJob,
    DownloaderJobOriginalFileAssociation,
    ProcessorJobOriginalFileAssociation,
    ProcessorJobDatasetAssociation,
    SurveyJobKeyValue
)
from data_refinery_common.message_queue import send_job
from data_refinery_common.job_lookup import (
    Downloaders,
    ProcessorPipeline,
    SurveyJobTypes,
    does_processor_job_have_samples,
    is_file_rnaseq,
)
from data_refinery_common.logging import get_and_configure_logger
from data_refinery_common.utils import (
    get_active_volumes,
    get_env_variable,
    get_env_variable_gracefully
)


logger = get_and_configure_logger(__name__)

# Maximum number of retries, so the number of attempts will be one
# greater than this because of the first attempt
MAX_NUM_RETRIES = 2

# This can be overritten by the env var "MAX_TOTAL_JOBS"
DEFAULT_MAX_JOBS = 20000

# The fastest each thread will repeat its checks.
# Could be slower if the thread takes longer than this to check its jobs.
MIN_LOOP_TIME = timedelta(minutes=2)

# The amount of time the main loop will wait in between checking if
# threads are still alive and then heart beating.
THREAD_WAIT_TIME = timedelta(minutes=10)

# How frequently we dispatch Janitor jobs.
JANITOR_DISPATCH_TIME = timedelta(minutes=30)

# How frequently we clean unplaceable jobs out of the Nomad queue.
CLEANUP_QUEUE_TIME = timedelta(minutes=30)


def read_config_list(config_file: str) -> List[str]:
    """
    Reads a file and returns a list with one item per line.
    """
    return_list = []
    with open(config_file) as config_list:
        for line in config_list:
            return_list.append(line.strip())

    return return_list


# These are lists of accessions that are for
# subsets of experiments that we want to prioritize.
PEDIATRIC_ACCESSION_LIST = read_config_list("config/pediatric_accessions.txt")
HGU133PLUS2_ACCESSION_LIST = read_config_list("config/hgu133plus2_accessions.txt")


##
# Utilities
##

def do_forever(min_loop_time: timedelta) -> Callable:
    """Run the wrapped function in a loop forever.

    The function won't be run more often than once per min_loop_time,
    however if it takes longer to run than min_loop_time, then it will
    be run less often than once per min_loop_time.
    """
    def decorator(function: Callable) -> Callable:
        @wraps(function)
        def wrapper(*args, **kwargs):
            while(True):
                start_time = timezone.now()

                try:
                    function(*args, **kwargs)
                except:
                    logger.exception("Exception caught by Foreman while running " + function.__name__)

                loop_time = timezone.now() - start_time
                if loop_time < min_loop_time:
                    remaining_time = min_loop_time - loop_time
                    time.sleep(remaining_time.seconds)

        return wrapper
    return decorator


def handle_repeated_failure(job) -> None:
    """If a job fails too many times, log it and stop retrying."""
    # Not strictly retried but will prevent the job from getting
    # retried any more times.
    job.retried = True

    # success may already be False, but if it was a hung or lost job
    # this will ensure it's marked as failed.
    job.success = False
    job.save()

    # At some point this should become more noisy/attention
    # grabbing. However for the time being just logging should be
    # sufficient because all log messages will be closely monitored
    # during early testing stages.
    logger.warn("%s #%d failed %d times!!!", job.__class__.__name__, job.id, MAX_NUM_RETRIES + 1)


##
# Job Prioritization
##


def prioritize_salmon_jobs(jobs: List) -> List:
    """Prioritizes salmon experiments based on how close to completion they are.

    This is because salmon experiments have a final processing step
    that must be performed on all the samples in the experiment, so if
    9/10 samples in an experiment are processed then they can't
    actually be used until that last sample is processed.
    """
    # The strategy for doing so is to build a mapping between every
    # salmon job in `jobs` to a priority. This priority will be what
    # percentage of the samples in this experiment have been
    # processed. Once we have that mapping we can sort those jobs by
    # that priority and move them to the front of the list.
    prioritized_jobs = []
    for job in jobs:
        try:
            if not does_processor_job_have_samples(job):
                continue

            # Salmon jobs are specifc to one sample.
            sample = job.get_samples().pop()

            # Skip jobs that aren't for Salmon. Handle both ProcessorJobs and DownloaderJobs.
            if type(job) is ProcessorJob and job.pipeline_applied != ProcessorPipeline.SALMON.value:
                continue
            elif type(job) is DownloaderJob:
                is_salmon_sample = False
                for original_file in sample.original_files.all():
                    if is_file_rnaseq(original_file.filename):
                        is_salmon_sample = True

                if not is_salmon_sample:
                    continue

            # Get a set of unique samples that share at least one
            # experiment with the sample this job is for.
            related_samples = set()
            for experiment in sample.experiments.all():
                for related_sample in experiment.samples.all():
                    related_samples.add(related_sample)

            # We cannot simply filter on is_processed because that field
            # doesn't get set until every sample in an experiment is processed.
            # Instead we are looking for one successful processor job.
            processed_samples = 0
            for related_sample in related_samples:
                original_files = related_sample.original_files
                if original_files.count() == 0:
                    logger.error("Salmon sample found without any original files!!!", sample=related_sample)
                elif original_files.first().processor_jobs.filter(success=True).count() >= 1:
                    processed_samples += 1

            experiment_completion_percent = processed_samples / len(related_samples)
            prioritized_jobs.append({"job": job, "priority": experiment_completion_percent})
        except:
            logger.debug("Exception caught while prioritizing salmon jobs!", job=job)


    sorted_job_mappings = sorted(prioritized_jobs, reverse=True, key=lambda k: k["priority"])
    sorted_jobs = [job_mapping["job"] for job_mapping in sorted_job_mappings]

    # Remove all the jobs we're moving to the front of the list
    for job in sorted_jobs:
        jobs.remove(job)

    return sorted_jobs + jobs


def prioritize_zebrafish_jobs(jobs: List) -> List:
    """Moves zebrafish jobs to the beginnging of the input list."""
    zebrafish_jobs = []
    for job in jobs:
        try:
            if not does_processor_job_have_samples(job):
                continue

            # There aren't cross-species jobs, so just checking one sample's organism will be sufficient.
            samples = job.get_samples()

            for sample in samples:
                if sample.organism.name == 'DANIO_RERIO':
                    zebrafish_jobs.append(job)
                    break
        except:
            logger.debug("Exception caught while prioritizing zebrafish jobs!", job=job)

    # Remove all the jobs we're moving to the front of the list
    for job in zebrafish_jobs:
        jobs.remove(job)

    return zebrafish_jobs + jobs


def prioritize_jobs_by_accession(jobs: List, accession_list: List[str]) -> List:
    """Moves jobs whose accessions are in accession_lst to the beginning of the input list."""
    prioritized_jobs = []
    for job in jobs:
        try:
            if not does_processor_job_have_samples(job):
                continue

            # All samples in a job correspond to the same experiment, so just check one sample.
            samples = job.get_samples()

            # Iterate through all the samples' experiments until one is
            # found with an accession in `accession_list`.
            is_prioritized_job = False
            for sample in samples:
                if is_prioritized_job:
                    # We found one! So stop looping
                    break

                for experiment in sample.experiments.all():
                    if experiment.accession_code in accession_list:
                        prioritized_jobs.append(job)
                        is_prioritized_job = True
                        break
        except:
            logger.exception("Exception caught while prioritizing jobs by accession!", job=job)

    # Remove all the jobs we're moving to the front of the list
    for job in prioritized_jobs:
        jobs.remove(job)

    return prioritized_jobs + jobs


##
# Downloaders
##


@retry(stop_max_attempt_number=3)
@transaction.atomic
def requeue_downloader_job(last_job: DownloaderJob) -> None:
    """Queues a new downloader job.

    The new downloader job will have num_retries one greater than
    last_job.num_retries.
    """
    num_retries = last_job.num_retries + 1

    new_job = DownloaderJob(num_retries=num_retries,
                            downloader_task=last_job.downloader_task,
                            accession_code=last_job.accession_code)
    new_job.save()

    for original_file in last_job.original_files.all():
        DownloaderJobOriginalFileAssociation.objects.get_or_create(downloader_job=new_job,
                                                           original_file=original_file)

    logger.debug("Requeuing Downloader Job which had ID %d with a new Downloader Job with ID %d.",
                 last_job.id,
                 new_job.id)
    try:
        if send_job(Downloaders[last_job.downloader_task], job=new_job, is_dispatch=True):
            last_job.retried = True
            last_job.success = False
            last_job.retried_job = new_job
            last_job.save()
        else:
            # Can't communicate with nomad just now, leave the job for a later loop.
            new_job.delete()
    except:
        logger.error("Failed to requeue Downloader Job which had ID %d with a new Downloader Job with ID %d.",
                     last_job.id,
                     new_job.id)
        # Can't communicate with nomad just now, leave the job for a later loop.
        new_job.delete()


def handle_downloader_jobs(jobs: List[DownloaderJob]) -> None:
    """For each job in jobs, either retry it or log it."""

    nomad_host = get_env_variable("NOMAD_HOST")
    nomad_port = get_env_variable("NOMAD_PORT", "4646")
    nomad_client = Nomad(nomad_host, port=int(nomad_port), timeout=30)
    # Maximum number of total jobs running at a time.
    # We do this now rather than import time for testing purposes.
    MAX_TOTAL_JOBS = int(get_env_variable_gracefully("MAX_TOTAL_JOBS", DEFAULT_MAX_JOBS))
    len_all_jobs = len(nomad_client.jobs.get_jobs())
    if len_all_jobs >= MAX_TOTAL_JOBS:
        logger.info("Not requeuing job until we're running fewer jobs.")
        return False

    # We want zebrafish data first, then hgu133plus2, then data
    # related to pediatric cancer, then to finish salmon experiments
    # that are close to completion.
    # Each function moves the jobs it prioritizes to the front of the
    # list, so apply them in backwards order.
    # jobs = prioritize_salmon_jobs(jobs)
    # jobs = prioritize_jobs_by_accession(jobs, PEDIATRIC_ACCESSION_LIST)
    # jobs = prioritize_jobs_by_accession(jobs, HGU133PLUS2_ACCESSION_LIST)
    # jobs = prioritize_zebrafish_jobs(jobs)

    jobs_dispatched = 0
    for count, job in enumerate(jobs):
        if job.num_retries < MAX_NUM_RETRIES:
            requeue_downloader_job(job)
            jobs_dispatched = jobs_dispatched + 1
        else:
            handle_repeated_failure(job)

        if (count % 100) == 0:
            len_all_jobs = len(nomad_client.jobs.get_jobs())

        if (jobs_dispatched + len_all_jobs) >= MAX_TOTAL_JOBS:
            logger.info("We hit the maximum total jobs ceiling, so we're not handling any more downloader jobs now.")
            return False

    return True

@do_forever(MIN_LOOP_TIME)
def retry_failed_downloader_jobs() -> None:
    """Handle downloader jobs that were marked as a failure."""
    failed_jobs = DownloaderJob.objects.filter(
        success=False,
        retried=False
    ).prefetch_related(
        "original_files__samples"
    )
    failed_jobs_list = [job for job in failed_jobs]

    if failed_jobs_list:
        logger.info(
            "Handling failed (explicitly-marked-as-failure) jobs!",
            jobs_count=len(failed_jobs_list)
        )
        handle_downloader_jobs(failed_jobs_list)


@do_forever(MIN_LOOP_TIME)
def retry_hung_downloader_jobs() -> None:
    """Retry downloader jobs that were started but never finished."""
    potentially_hung_jobs = DownloaderJob.objects.filter(
        success=None,
        retried=False,
        end_time=None,
        start_time__isnull=False,
        no_retry=False
    ).prefetch_related(
        "original_files__samples"
    )

    nomad_host = get_env_variable("NOMAD_HOST")
    nomad_port = get_env_variable("NOMAD_PORT", "4646")
    nomad_client = Nomad(nomad_host, port=int(nomad_port), timeout=30)
    hung_jobs = []
    for job in potentially_hung_jobs:
        try:
            job_status = nomad_client.job.get_job(job.nomad_job_id)["Status"]
            if job_status != "running":
                # Make sure it didn't finish since our original query.
                job.refresh_from_db()
                if job.end_time is None:
                    hung_jobs.append(job)
        except URLNotFoundNomadException:
            hung_jobs.append(job)
        except Exception:
            logger.exception("Couldn't query Nomad about Processor Job.", processor_job=job.id)

    handle_downloader_jobs(hung_jobs)


@do_forever(MIN_LOOP_TIME)
def retry_lost_downloader_jobs() -> None:
    """Retry downloader jobs that went too long without being started.

    Idea: at some point this function could integrate with the spot
    instances to determine if jobs are hanging due to a lack of
    instances. A naive time-based implementation like this could end
    up retrying every single queued job if there were a long period
    during which the price of spot instance is higher than our bid
    price.
    """
    potentially_lost_jobs = DownloaderJob.objects.filter(
        success=None,
        retried=False,
        start_time=None,
        end_time=None,
        no_retry=False
    ).prefetch_related(
        "original_files__samples"
    )

    nomad_host = get_env_variable("NOMAD_HOST")
    nomad_port = get_env_variable("NOMAD_PORT", "4646")
    nomad_client = Nomad(nomad_host, port=int(nomad_port), timeout=30)
    lost_jobs = []
    for job in potentially_lost_jobs:
        try:
            if job.nomad_job_id:
                job_status = nomad_client.job.get_job(job.nomad_job_id)["Status"]
                # If the job is still pending, then it makes sense that it
                # hasn't started and if it's running then it may not have
                # been able to mark the job record as started yet.
                if job_status != "pending" and job_status != "running":
                    logger.debug(("Determined that a downloader job needs to be requeued because its"
                                  " Nomad Job's status is: %s."),
                                 job_status,
                                 job_id=job.id
                    )
                    lost_jobs.append(job)
            else:
                # If there is no nomad_job_id field set, we could be
                # in the small window where the job was created but
                # hasn't yet gotten a chance to be queued.
                # If this job really should be restarted we'll get it in the next loop.
                if timezone.now() - job.created_at > MIN_LOOP_TIME:
                    lost_jobs.append(job)
        except socket.timeout:
            logger.info("Timeout connecting to Nomad - is Nomad down?", job_id=job.id)
        except URLNotFoundNomadException:
            logger.debug(("Determined that a downloader job needs to be requeued because "
                              "querying for its Nomad job failed: "),
                             job_id=job.id
            )
            lost_jobs.append(job)
        except nomad.api.exceptions.BaseNomadException:
            logger.info("Problem connecting to Nomad - is Nomad down?", job_id=job.id)
        except Exception:
            logger.exception("Couldn't query Nomad about Processor Job.", processor_job=job.id)

    handle_downloader_jobs(lost_jobs)


##
# Processors
##


@retry(stop_max_attempt_number=3)
@transaction.atomic
def requeue_processor_job(last_job: ProcessorJob) -> None:
    """Queues a new processor job.

    The new processor job will have num_retries one greater than
    last_job.num_retries.
    """
    num_retries = last_job.num_retries + 1

    # The Salmon pipeline is quite RAM-sensitive.
    # Try it again with an increased RAM amount, if possible.
    new_ram_amount = last_job.ram_amount
    if last_job.pipeline_applied == "SALMON":
        if new_ram_amount == 4096:
            new_ram_amount = 8192
        elif new_ram_amount == 8192:
            new_ram_amount = 12288
        elif new_ram_amount == 12288:
            new_ram_amount = 16384

    # The AFFY pipeline is somewhat RAM-sensitive.
    # Try it again with an increased RAM amount, if possible.
    new_ram_amount = last_job.ram_amount
    if last_job.pipeline_applied == "AFFY_TO_PCL":
        if new_ram_amount == 2048:
            new_ram_amount = 4096
        elif new_ram_amount == 4096:
            new_ram_amount = 8192

    new_job = ProcessorJob(num_retries=num_retries,
                           pipeline_applied=last_job.pipeline_applied,
                           ram_amount=new_ram_amount,
                           volume_index=last_job.volume_index)
    new_job.save()

    for original_file in last_job.original_files.all():
        ProcessorJobOriginalFileAssociation.objects.get_or_create(processor_job=new_job,
                                                          original_file=original_file)

    for dataset in last_job.datasets.all():
        ProcessorJobDatasetAssociation.objects.get_or_create(processor_job=new_job,
                                                     dataset=dataset)

    try:
        logger.debug("Requeuing Processor Job which had ID %d with a new Processor Job with ID %d.",
                     last_job.id,
                     new_job.id)
        if send_job(ProcessorPipeline[last_job.pipeline_applied], job=new_job, is_dispatch=True):
            last_job.retried = True
            last_job.success = False
            last_job.retried_job = new_job
{'fraction_non_alphanumeric': 0.06654964117920388, 'fraction_numerical': 0.00516288915276989, 'mean_word_length': 2.4158363460012344, 'pattern_counts': {'":': 11, '<': 7, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 96, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '924826', 'n_tokens_mistral': 11309, 'n_tokens_neox': 10569, 'n_words': 3327}