<commit_before>import config from './config';
import logger from './utils/logger';

import Queue from 'bull';
import { getStatsDClient } from './utils/statsd';

const queueSettings = {
    lockDuration: 60000,
    lockRenewTime: 15000,
    stalledInterval: 75000,
    maxStalledCount: 2,
    guardInterval: 3500,
};

export const rssQueue = new Queue('rss', config.cache.uri, {
	settings: queueSettings,
});
export const ogQueue = new Queue('og', config.cache.uri, {
	settings: queueSettings,
});
export const podcastQueue = new Queue('podcast', config.cache.uri, {
	settings: queueSettings,
});
export const streamQueue = new Queue('stream', config.cache.uri, {
	limiter: { max: 12000, duration: 3600000 }, // 12k per hour
	settings: queueSettings,
});
export const socialQueue = new Queue('social', config.cache.uri, {
	settings: queueSettings,
});

function makeMetricKey(queue, event) {
	return ['winds', 'bull', queue.name, event].join('.');
}

async function trackQueueSize(statsd, queue) {
	let queueStatus = await queue.getJobCounts();
	statsd.gauge(makeMetricKey(queue, 'waiting'), queueStatus.waiting);
	statsd.gauge(makeMetricKey(queue, 'active'), queueStatus.active);
}

function AddQueueTracking(queue) {
	var statsd = getStatsDClient();

	queue.on('error', function(err) {
		statsd.increment(makeMetricKey(queue, 'error'));
		logger.warn(`Queue ${queue.name} encountered an unexpected error: ${err.message}`);
	});

	queue.on('active', function(job, jobPromise) {
		statsd.increment(makeMetricKey(queue, 'active'));
	});

	queue.on('completed', function(job, result) {
		statsd.timing(makeMetricKey(queue, 'elapsed'), new Date() - job.timestamp);
		statsd.increment(makeMetricKey(queue, 'completed'));
	});

	queue.on('failed', function(job, err) {
		statsd.increment(makeMetricKey(queue, 'failed'));
		logger.warn(`Queue ${queue.name} failed to process job '${JSON.stringify(job)}': ${err.message}`);
	});

	queue.on('paused', function() {
		statsd.increment(makeMetricKey(queue, 'paused'));
	});

	queue.on('resumed', function(job) {<commit_msg>Print out stalled jobs as soon as it happens<commit_after>import config from './config';
import logger from './utils/logger';

import Queue from 'bull';
import { getStatsDClient } from './utils/statsd';

const queueSettings = {
    lockDuration: 60000,
    lockRenewTime: 15000,
    stalledInterval: 75000,
    maxStalledCount: 2,
    guardInterval: 3500,
};

export const rssQueue = new Queue('rss', config.cache.uri, {
	settings: queueSettings,
});
export const ogQueue = new Queue('og', config.cache.uri, {
	settings: queueSettings,
});
export const podcastQueue = new Queue('podcast', config.cache.uri, {
	settings: queueSettings,
});
export const streamQueue = new Queue('stream', config.cache.uri, {
	limiter: { max: 12000, duration: 3600000 }, // 12k per hour
	settings: queueSettings,
});
export const socialQueue = new Queue('social', config.cache.uri, {
	settings: queueSettings,
});

function makeMetricKey(queue, event) {
	return ['winds', 'bull', queue.name, event].join('.');
}

async function trackQueueSize(statsd, queue) {
	let queueStatus = await queue.getJobCounts();
	statsd.gauge(makeMetricKey(queue, 'waiting'), queueStatus.waiting);
	statsd.gauge(makeMetricKey(queue, 'active'), queueStatus.active);
}

function AddQueueTracking(queue) {
	var statsd = getStatsDClient();

	queue.on('error', function(err) {
		statsd.increment(makeMetricKey(queue, 'error'));
		logger.warn(`Queue ${queue.name} encountered an unexpected error: ${err.message}`);
	});

	queue.on('active', function(job, jobPromise) {
		statsd.increment(makeMetricKey(queue, 'active'));
	});

	queue.on('completed', function(job, result) {
		statsd.timing(makeMetricKey(queue, 'elapsed'), new Date() - job.timestamp);
		statsd.increment(makeMetricKey(queue, 'completed'));
	});

	queue.on('stalled', function(job) {
		statsd.increment(makeMetricKey(queue, 'stalled'));
		logger.warn(`Queue ${queue.name} job stalled: '${JSON.stringify(job)}'`);
	});

	queue.on('failed', function(job, err) {
		statsd.increment(makeMetricKey(queue, 'failed'));
		logger.warn(`Queue ${queue.name} failed to process job '${JSON.stringify(job)}': ${err.message}`);
	});

	queue.on('paused', function() {
		statsd.increment(makeMetricKey(queue, 'paused'));
	});

	queue.on('resumed', function(job) {
{'fraction_non_alphanumeric': 0.16716763005780347, 'fraction_numerical': 0.015722543352601155, 'mean_word_length': 5.675925925925926, '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': '3059666', 'n_tokens_mistral': 1494, 'n_tokens_neox': 1393, 'n_words': 313}