5292902236

Build Type

push

github

Committed by Leandro Felicio

Commit Message

Adding new worker metrics

Run Details

343 of 367 branches covered (93.46%)

Branch coverage included in aggregate %.

17 of 17 new or added lines in 3 files covered. (100.0%)

4 existing lines in 1 file now uncovered.

448 of 536 relevant lines covered (83.58%)

4.18 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.21

/remotecv/unique_queue.py

from pyres import ResQ
from pyres.worker import Worker

from remotecv.timing import get_time, get_interval
from remotecv.utils import context, logger


class UniqueQueue(ResQ):
    def _escape_for_key(self, value):
        return value.replace(" ", "").replace("\n", "")

    def _create_unique_key(self, queue, key):
        return f"resque:unique:queue:{queue}:{self._escape_for_key(str(key))}"

    def add_unique_key(self, queue, key):
        unique_key = self._create_unique_key(queue, key)
        if self.redis.get(unique_key) == b"1":
            # Do nothing as this message is already enqueued
            return False
        self.redis.set(unique_key, "1")

        return True

    def del_unique_key(self, queue, key):
        start_time = get_time()
        unique_key = self._create_unique_key(queue, key)
        self.redis.delete(unique_key)
        context.metrics.timing(
            "worker.del_unique_key.time",
            get_interval(start_time, get_time()),
        )

    def enqueue_unique_from_string(
        self, klass_as_string, queue, args=None, key=None
    ):
        if not self.add_unique_key(queue, key):
            logger.debug("key %s already enqueued", key)
            return

        if not args:
            args = []

        payload = {
            "class": klass_as_string,
            "queue": queue,
            "args": args + [key],
            "key": key,
        }
        self.push(queue, payload)
        logger.info("enqueued '%s' job on queue %s", klass_as_string, queue)
        if args:
            logger.debug("job arguments: %s", str(args))
        else:
            logger.debug("no arguments passed in.")


class UniqueWorker(Worker):
    def __init__(
        self,
        queues=(),
        server="localhost:6379",
        password=None,
        timeout=None,
        after_fork=None,
    ):
        self.after_fork_fn = after_fork
        super().__init__(queues, UniqueQueue(server=server), password, timeout)

    def after_fork(self, job):
        if self.after_fork_fn and callable(self.after_fork_fn):
            self.after_fork_fn(job)

    def before_process(self, job):
        self.resq.del_unique_key(
            job._queue, job._payload["key"]  # pylint: disable=protected-access
        )
        return job

    def reserve(self, timeout=10):
        start_time = get_time()
        job = super().reserve(timeout=timeout)
        context.metrics.timing(
            "worker.read_queue.time",
            get_interval(start_time, get_time()),
        )
        return job

1	from pyres import ResQ	5✔
2	from pyres.worker import Worker	5✔
3
4	from remotecv.timing import get_time, get_interval	5✔
5	from remotecv.utils import context, logger	5✔
6
7
8	class UniqueQueue(ResQ):	5✔
9	def _escape_for_key(self, value):	5✔
10	return value.replace(" ", "").replace("\n", "")	5✔
11
12	def _create_unique_key(self, queue, key):	5✔
13	return f"resque:unique:queue:{queue}:{self._escape_for_key(str(key))}"	5✔
14
15	def add_unique_key(self, queue, key):	5✔
16	unique_key = self._create_unique_key(queue, key)	5✔
17	if self.redis.get(unique_key) == b"1":	5✔
18	# Do nothing as this message is already enqueued
19	return False	5✔
20	self.redis.set(unique_key, "1")	5✔
21
22	return True	5✔
23
24	def del_unique_key(self, queue, key):	5✔
25	start_time = get_time()	5✔
26	unique_key = self._create_unique_key(queue, key)	5✔
27	self.redis.delete(unique_key)	5✔
28	context.metrics.timing(	5✔
29	"worker.del_unique_key.time",
30	get_interval(start_time, get_time()),
31	)
32
33	def enqueue_unique_from_string(	5✔
34	self, klass_as_string, queue, args=None, key=None
35	):
36	if not self.add_unique_key(queue, key):	5✔
37	logger.debug("key %s already enqueued", key)	5✔
38	return	5✔
39
40	if not args:	5✔
41	args = []	5✔
42
43	payload = {	5!
44	"class": klass_as_string,
45	"queue": queue,
46	"args": args + [key],
47	"key": key,
48	}
49	self.push(queue, payload)	5✔
50	logger.info("enqueued '%s' job on queue %s", klass_as_string, queue)	5✔
51	if args:	5✔
UNCOV 52	logger.debug("job arguments: %s", str(args))	×
53	else:
54	logger.debug("no arguments passed in.")	5✔
55
56
57	class UniqueWorker(Worker):	5✔
58	def __init__(	5✔
59	self,
60	queues=(),
61	server="localhost:6379",
62	password=None,
63	timeout=None,
64	after_fork=None,
65	):
66	self.after_fork_fn = after_fork	5✔
67	super().__init__(queues, UniqueQueue(server=server), password, timeout)	5✔
68
69	def after_fork(self, job):	5✔
UNCOV 70	if self.after_fork_fn and callable(self.after_fork_fn):	×
UNCOV 71	self.after_fork_fn(job)	×
72
73	def before_process(self, job):	5✔
UNCOV 74	self.resq.del_unique_key(	×
75	job._queue, job._payload["key"] # pylint: disable=protected-access
76	)
77	return job	×
78
79	def reserve(self, timeout=10):	5✔
80	start_time = get_time()	5✔
81	job = super().reserve(timeout=timeout)	5✔
82	context.metrics.timing(	5✔
83	"worker.read_queue.time",
84	get_interval(start_time, get_time()),
85	)
86	return job	5✔

thumbor / remotecv / 5292902236

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous