Separate Task/Queue for SearchIndex

[jobelenus]

I have a lot of real time indexing going on in my application. There is one model that actually has a higher priority of being indexed. But these are a minority of indexing jobs going on in the background. Other jobs are ahead of it in the queue so it sits there waiting.

I can solve this problem by creating a separate route for a specific task of my own creation that will extend your task and do nothing different. The problem is how to get the signal processor to launch that task instead of your default task. I see the optional task path arg and the setting for the task. But the problem is that in enqueue doesn't pass an argument.

I was hoping to avoid copying that method in my own signal processor just to change one line for passing an argument. Could there be a class attribute on the signal processor that represents the task path?

Or do you have a better proposal for solving this specific problem. I can of course actually do the work and submit a PR - just hoping to get some communication here that this is the desired approach and that the PR would be accepted and get released fairly quickly?

[SteveByerly]

@jobelenus I don't have any experience using celery-haystack, but I have plenty using celery. I handle different priority queues in some of my own applications, so I can't imagine it being a huge problem here - I just need to know where to start looking. Can you point me to where the current bottleneck is and where you'd expect the change to take place?

[jobelenus]

Sure. Starting in the signal processor: https://github.com/django-haystack/celery-haystack/blob/develop/celery_haystack/signals.py#L44

And this is the def enqueue_task: https://github.com/django-haystack/celery-haystack/blob/develop/celery_haystack/utils.py#L41

Where def get_update_task(task_path=None): is the method signature (that defaults to the setting: CELERY_HAYSTACK_DEFAULT_TASK).

Let me know if thats confusing

[jobelenus]

Since celery doesn't support message priority I have to separate out the messages into different queues and then allocate the amount of resources I want to each queue. Correct?

[SteveByerly]

Yes. You'll basically want to send your special indexes to a different queue(s) than the rest so they end up getting processed faster.

I've briefly looked at the code and I'm not sure the best approach.
One option would be to define which queue to use on a per-index basis, defaulting to the CELERY_HAYSTACK_QUEUE setting. Perhaps have an attribute or method on CelerySearchIndex that is retrieved in the signal's enqueue_task and passed into enqueue_task as a kwarg.

What were you thinking your approach would be?

[jobelenus]

That is precisely what I was thinking. More than happy to do the work. Just
wanted to make sure thats an approach you all want to support/accept into
the project

On Wed, Jul 27, 2016 at 6:08 PM, Steve Byerly notifications@github.com
wrote:

Yes. You'll basically want to send your special indexes to a different
queue(s) than the rest so they end up getting processed faster.

I've briefly looked at the code and I'm not sure the best approach.
One option would be to define the queue to use on a per-index basis,
defaulting to the CELERY_HAYSTACK_QUEUE setting. Perhaps have an
attribute or method on CelerySearchIndex that is retrieved in the
signal's enqueue_task and passed into enqueue_task as a kwarg.

What were you thinking your approach would be?

—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub
#59 (comment),
or mute the thread
https://github.com/notifications/unsubscribe-auth/AAEPpSfJ_t41ISq8okCevLY6ZQSLonntks5qZ9b5gaJpZM4JWUti
.

[SteveByerly]

I think it's a reasonable addition. Go for it.

[jobelenus]

Hrmm, this is odd... while testing I'm getting this trace. I've forked off master and made my changes, but something seems to be not quite write with the app config? The celery_haystack is still in INSTALLED_APPS, I'm not sure what I'm missing here:

File "/srv/lenspro/env/local/lib/python2.7/site-packages/celery_haystack/utils.py", line 36, in enqueue_task
if settings.CELERY_HAYSTACK_QUEUE:
out: AttributeError: 'Settings' object has no attribute 'CELERY_HAYSTACK_QUEUE'

Edit: I thought I figured out what was going on and imported the celery_haystack conf settings in my own application task, but it didn't quite do the trick:

from celery_haystack.tasks import CeleryHaystackSignalHandler
from celery_haystack.conf import settings
class CustomerCeleryHaystackSignalHandler(CeleryHaystackSignalHandler):
    pass

I can see that CeleryHaystackSignalHandler.using is filled in, but the setting it is filled in with (settings.CELERY_HAYSTACK_DEFAULT_ALIAS) doesn't exist. Where is the magic I am missing?

Edit2: Actually even without that settings import my own task CustomerCeleryHaystackSignalHandler.using is filled in. I wonder what would have changed that the utils.py file now has bad settings? I don't think it was anything I changed, you can see my commit here: jobelenus@425f5be

I happen to be in Django 1.9.6 if that makes a difference, I see the travis config has 1.9

[SteveByerly]

It looks like an issue with app loading and your new tasks. When i run celery_haystack locally, I can import the settings fine.

If you add CELERY_HAYSTACK_QUEUE setting in your own settings, what happens? If it works fine, then I think something needs to be loaded/moved.

[jobelenus]

I can do from celery_haystack.conf import settings but that settings dict doesn't have any CELERY_X.... settings in it. Where is the magic that loads the CeleryHaystack AppConf into the settings? Because that is what it seems I am missing/cannot find/don't see anywhere else in the project.

When I had celery_haystack installed via pip from pypi there wasn't a problem. I branched from master, is there something on master that's wrong and I should be using something else?

[SteveByerly]

I believe it gets loaded when django does its magic app loading. I believe Celery also does some loading of tasks too.

What happens when you manually add the fully qualified setting (CELERY_HAYSTACK_QUEUE) to conf.py? Does it work/any new error?

[jobelenus]

Well of course that will work, if i add the setting -- but that isn't really a solution to the problem, and there are at least 3 settings that the utils.py file needs. What is really weird is that the indexes.py file seems to work fine with the app loading, but not utils. Thats what confuses me the most about this

[jobelenus]

I thought perhaps this would fix it, but it doesn't jobelenus@2479424 (e.g. that the relative import was using my apps settings file instead of the celery_haystack)

[SteveByerly]

Can you post the new code you added to your own project? Either here or a gist?

[jobelenus]

Sure, its really not much. I'm not even using the new feature yet.

lenspro.tasks.py:

+from celery_haystack.tasks import CeleryHaystackSignalHandler
+class CustomerCeleryHaystackSignalHandler(CeleryHaystackSignalHandler):
+    pass

lenspro.settings.py:

CELERY_ROUTES = {
...
+    'lenspro.tasks.CustomerCeleryHaystackSignalHandler': {'queue': 'customerindex'}
}

lenspro.search_indexes.py: this already existed

class CustomerIndex(CelerySearchIndex, indexes.Indexable):
    text = indexes.CharField(document=True, use_template=True)
    customer_id = indexes.IntegerField(model_attr='id', null=False)

    def get_model(self):
        return User

When I create a new User is when I see the error. I didn't even put the task_path attribute into my Index. So there is something going on here. I'm installing it with my requirements.txt file with this entry git+https://github.com/jobelenus/celery-haystack.git@searchindex-taskpath

[SteveByerly]

Why are you using a custom signal handler if the goal is to define the task queues on the searchindex?

Then instead of defining the routes, you can define the queues, though i don't think you have to - they'll get added by celery auto-magically:

CELERY_QUEUES = (
    Queue('default', Exchange('default', durable=True), routing_key='default'),
    Queue('priority', Exchange('priority', durable=True), routing_key='priority'),
)