use Redis for custom plugin

[Linkk93]

Hi all,

I am developing a custom plugin and want to use Redis job queuing for long background tasks.
I saw that by default the queues high default low are started by the netbox-rq service.
I tried to enqueue to the default queue but the job is never started.

Status: queued
Job ID: bea1e034-e820-4de8-ac5c-a0279336e861
Job started_at: None

After investigating a bit, I noticed that the netbox-rq service is stopped
(code=exited, status=1/FAILURE)
That is not correct, is it?

Is it generally good practice to use the netbox-rq for custom plugins?

The code I currently test with is the following, maybe I am missing something:

redis = Redis()
rq_queue = Queue(connection=redis)
job = rq_queue.enqueue(sync_ap,
                                   kwargs={
                                       'macaddr': local_ap.mac_address,
                                       'serial': local_ap.serial
                                   })
workers = Worker.all(connection=redis)
print(f'workers: {workers}')
print('Status: %s' % job.get_status())
print(f'Job ID: {job.id}')
print(f'Job started_at: {job.started_at}')
time.sleep(5)
print(f'workers: {workers}')
print('Status: %s' % job.get_status())
print(f'Job ID: {job.id}')
print(f'Job started_at: {job.started_at}')
time.sleep(5)
print('Status: %s' % job.get_status())
print(f'Job ID: {job.id}')
print(f'Job started_at: {job.started_at}')

I am still very new to Redis, so every help is appreciated.

[kkthxbye-code]

After investigating a bit, I noticed that the netbox-rq service is stopped
(code=exited, status=1/FAILURE)
That is not correct, is it?

rqworker needs to be running to use rq.

Is it generally good practice to use the netbox-rq for custom plugins?

Yep, it's supported.

Your code is funky though, so not sure it will work.

Something like this all that is needed:

def my_func(kwarg_test, kwarg_blah):
    print(kwarg_test, kwarg_blah)

queue = django_rq.get_queue("high")
queue.enqueue(
    my_func,
    kwarg_test="test",
    kwarg_blah="blah",
)

[Linkk93]

Thanks for the help.
I figured out what was wrong with the rqworker service and it's fixed now.

OK, so I'm using the django_rq library now, like in your example.
For the moment I have also removed the arguments to keep it simple.

import django_rq

def sync_ap():
    print('doing a long task')

def function_called_in_view():
    default_queue = django_rq.get_queue('default')
    local_job= default_queue.enqueue(sync_ap)
    
    print('Status: %s' % local_job.get_status())
    print(f'Job ID: {local_job.id}')
    print(f'Job started_at: {local_job.started_at}')
    time.sleep(5)
    print('Status: %s' % local_job.get_status())
    print(f'Job ID: {local_job.id}')
    print(f'Job started_at: {local_job.started_at}')

But it will now fail:

Status: queued
Job ID: 575ec5c0-a857-47b9-99cb-b5b809466355
Job started_at: None
Status: failed
Job ID: 575ec5c0-a857-47b9-99cb-b5b809466355
Job started_at: None

I wonder why it fails when it was apparently never started

[kkthxbye-code]

No idea, you can check the django-rq dashboard in the admin panel under background tasks. It should list the failed task.

[Linkk93]

Thank you, I did not know about that!

[Linkk93]

Edit: nevermind, I had to give my function with the f=my_func attribute in enqueue

def my_func(kwarg_test, kwarg_blah):
    print(kwarg_test, kwarg_blah)

queue = django_rq.get_queue("high")
queue.enqueue(
    f=my_func,
    kwarg_test="test",
    kwarg_blah="blah",
)

Old post:

So, I am using this inside my custom plugin and it seems like rq is trying to get an attribute for it?

Traceback (most recent call last):
  File "/opt/netbox/venv/lib/python3.10/site-packages/rq/utils.py", line 118, in import_attribute
    attribute_owner = getattr(module, attribute_owner_name)
AttributeError: module 'my_custom_plugin.views' has no attribute ''

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/opt/netbox/venv/lib/python3.10/site-packages/rq/worker.py", line 1428, in perform_job
    rv = job.perform()
  File "/opt/netbox/venv/lib/python3.10/site-packages/rq/job.py", line 1278, in perform
    self._result = self._execute()
  File "/opt/netbox/venv/lib/python3.10/site-packages/rq/job.py", line 1315, in _execute
    result = self.func(*self.args, **self.kwargs)
  File "/opt/netbox/venv/lib/python3.10/site-packages/rq/job.py", line 425, in func
    return import_attribute(self.func_name)
  File "/opt/netbox/venv/lib/python3.10/site-packages/rq/utils.py", line 120, in import_attribute
    raise ValueError('Invalid attribute name: %s' % attribute_name)
ValueError: Invalid attribute name: print_job

Do I need to set this in my views.py?