<commit_before>package net.nemerosa.ontrack.job.support

import io.micrometer.core.instrument.MeterRegistry
import net.nemerosa.ontrack.common.Time
import net.nemerosa.ontrack.job.*
import org.apache.commons.lang3.Validate
import org.slf4j.LoggerFactory
import java.time.LocalDateTime
import java.time.temporal.ChronoUnit
import java.util.*
import java.util.concurrent.*
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicLong
import java.util.concurrent.atomic.AtomicReference
import java.util.function.BiFunction

/**
 * @property meterRegistry If set, the scheduler will register job metrics
 */
class DefaultJobScheduler
@JvmOverloads
constructor(
        private val jobDecorator: JobDecorator,
        private val schedulerPool: ScheduledExecutorService,
        private val jobListener: JobListener,
        initiallyPaused: Boolean,
        private val jobPoolProvider: BiFunction<ExecutorService, Job, ExecutorService>,
        private val scattering: Boolean,
        private val scatteringRatio: Double,
        private val meterRegistry: MeterRegistry? = null
) : JobScheduler {

    private val logger = LoggerFactory.getLogger(JobScheduler::class.java)

    private val services = ConcurrentHashMap(TreeMap<JobKey, JobScheduledService>())
    private val schedulerPaused: AtomicBoolean

    private val idGenerator = AtomicLong()

    @JvmOverloads
    constructor(
            jobDecorator: JobDecorator,
            schedulerPool: ScheduledExecutorService,
            jobListener: JobListener,
            initiallyPaused: Boolean,
            scattering: Boolean,
            scatteringRatio: Double,
            meterRegistry: MeterRegistry? = null
    ) : this(
            jobDecorator,
            schedulerPool,
            jobListener,
            initiallyPaused,
            BiFunction { executorService, _ -> executorService },
            scattering,
            scatteringRatio,
            meterRegistry
    )

    private fun MeterRegistry.statusGauge(
            name: String,
            statusFilterFn: (JobStatus) -> Boolean
    ) {
        gauge(
                "ontrack_job_${name}_total",
                services,
                {
                    it.filter { (_, service) -> statusFilterFn(service.jobStatus) }
                            .size.toDouble()
                }
        )
    }

    init {
        Validate.inclusiveBetween(0.0, 1.0, scatteringRatio)
        this.schedulerPaused = AtomicBoolean(initiallyPaused)
        // Metrics
        if (meterRegistry != null) {
            // count
            meterRegistry.gaugeMapSize(
                    "ontrack_job_count_total",
                    emptyList(),
                    services
            )
            meterRegistry.statusGauge("running", { it.isRunning })
            meterRegistry.statusGauge("disabled", { it.isDisabled })
            meterRegistry.statusGauge("paused", { it.isPaused })
            meterRegistry.statusGauge("error", { it.isError })
            meterRegistry.statusGauge("invalid", { !it.isValid })
            meterRegistry.gauge(
                    "ontrack_job_error_count_total",
                    services,
                    {
                        it.values.map { it.jobStatus.lastErrorCount }.sum().toDouble()
                    }
            )
        }
    }

    override fun schedule(job: Job, schedule: Schedule) {
        logger.info("[scheduler][job]{} Scheduling with {}", job.key, schedule)
        // Manages existing schedule
        val existingService = services[job.key]
        if (existingService != null) {
            logger.info("[scheduler][job]{} Modifying existing schedule", job.key)
            existingService.update(
                    job,
                    schedule
            )
        }
        // Creates and starts the scheduled service
        else {
            logger.info("[scheduler][job]{} Starting scheduled service", job.key)
            // Copy stats from old schedule
            val jobScheduledService = JobScheduledService(
                    job,
                    schedule,
                    schedulerPool,
                    jobListener.isPausedAtStartup(job.key)
            )
            // Registration
            services[job.key] = jobScheduledService
        }
    }

    override fun unschedule(key: JobKey): Boolean {
        return unschedule(key, true)<commit_msg>#726 Reduced logging<commit_after>package net.nemerosa.ontrack.job.support

import io.micrometer.core.instrument.MeterRegistry
import net.nemerosa.ontrack.common.Time
import net.nemerosa.ontrack.job.*
import org.apache.commons.lang3.Validate
import org.slf4j.LoggerFactory
import java.time.LocalDateTime
import java.time.temporal.ChronoUnit
import java.util.*
import java.util.concurrent.*
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicLong
import java.util.concurrent.atomic.AtomicReference
import java.util.function.BiFunction

/**
 * @property meterRegistry If set, the scheduler will register job metrics
 */
class DefaultJobScheduler
@JvmOverloads
constructor(
        private val jobDecorator: JobDecorator,
        private val schedulerPool: ScheduledExecutorService,
        private val jobListener: JobListener,
        initiallyPaused: Boolean,
        private val jobPoolProvider: BiFunction<ExecutorService, Job, ExecutorService>,
        private val scattering: Boolean,
        private val scatteringRatio: Double,
        private val meterRegistry: MeterRegistry? = null
) : JobScheduler {

    private val logger = LoggerFactory.getLogger(JobScheduler::class.java)

    private val services = ConcurrentHashMap(TreeMap<JobKey, JobScheduledService>())
    private val schedulerPaused: AtomicBoolean

    private val idGenerator = AtomicLong()

    @JvmOverloads
    constructor(
            jobDecorator: JobDecorator,
            schedulerPool: ScheduledExecutorService,
            jobListener: JobListener,
            initiallyPaused: Boolean,
            scattering: Boolean,
            scatteringRatio: Double,
            meterRegistry: MeterRegistry? = null
    ) : this(
            jobDecorator,
            schedulerPool,
            jobListener,
            initiallyPaused,
            BiFunction { executorService, _ -> executorService },
            scattering,
            scatteringRatio,
            meterRegistry
    )

    private fun MeterRegistry.statusGauge(
            name: String,
            statusFilterFn: (JobStatus) -> Boolean
    ) {
        gauge(
                "ontrack_job_${name}_total",
                services,
                {
                    it.filter { (_, service) -> statusFilterFn(service.jobStatus) }
                            .size.toDouble()
                }
        )
    }

    init {
        Validate.inclusiveBetween(0.0, 1.0, scatteringRatio)
        this.schedulerPaused = AtomicBoolean(initiallyPaused)
        // Metrics
        if (meterRegistry != null) {
            // count
            meterRegistry.gaugeMapSize(
                    "ontrack_job_count_total",
                    emptyList(),
                    services
            )
            meterRegistry.statusGauge("running", { it.isRunning })
            meterRegistry.statusGauge("disabled", { it.isDisabled })
            meterRegistry.statusGauge("paused", { it.isPaused })
            meterRegistry.statusGauge("error", { it.isError })
            meterRegistry.statusGauge("invalid", { !it.isValid })
            meterRegistry.gauge(
                    "ontrack_job_error_count_total",
                    services,
                    {
                        it.values.map { it.jobStatus.lastErrorCount }.sum().toDouble()
                    }
            )
        }
    }

    override fun schedule(job: Job, schedule: Schedule) {
        logger.debug("[scheduler][job]{} Scheduling with {}", job.key, schedule)
        // Manages existing schedule
        val existingService = services[job.key]
        if (existingService != null) {
            logger.debug("[scheduler][job]{} Modifying existing schedule", job.key)
            existingService.update(
                    job,
                    schedule
            )
        }
        // Creates and starts the scheduled service
        else {
            logger.debug("[scheduler][job]{} Starting scheduled service", job.key)
            // Copy stats from old schedule
            val jobScheduledService = JobScheduledService(
                    job,
                    schedule,
                    schedulerPool,
                    jobListener.isPausedAtStartup(job.key)
            )
            // Registration
            services[job.key] = jobScheduledService
        }
    }

    override fun unschedule(key: JobKey): Boolean {
        return unschedule(key, true)
{'fraction_non_alphanumeric': 0.08378439332431213, 'fraction_numerical': 0.0016914749661705007, 'mean_word_length': 1.9553482172609131, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 13, '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': '4048220', 'n_tokens_mistral': 2269, 'n_tokens_neox': 2201, 'n_words': 461}