Path: csiph.com!fu-berlin.de!uni-berlin.de!not-for-mail
From: nonami <nonami.ayo@gmail.com>
Newsgroups: comp.lang.python
Subject: Problems using celery and pyelasticsearch
Date: Mon, 7 Dec 2015 02:37:15 +0100
Lines: 73
Message-ID: <mailman.7.1449452239.12405.python-list@python.org>
Mime-Version: 1.0
Content-Type: text/plain; charset=utf-8; format=flowed
Content-Transfer-Encoding: 7bit
User-Agent: Mozilla/5.0 (X11; Linux x86_64; rv:38.0) Gecko/20100101 Thunderbird/38.4.0
Precedence: list
Xref: csiph.com comp.lang.python:100076

Hi,

I create a task that uses this function

def get_jobs(specialization, yoe):
     try:
         key = "SpecAlert-"
         if len(specialization):
             key += '-'.join(map(str, specialization))

         if yoe is not None:
             key += '-yoe'.join(yoe)

         jobs = memcache_client.get(key)

         if not jobs:
             spec_param = {'terms': {'jobs.specialization.id': 
specialization}}
             if yoe is not None:
                 if len(yoe) == 2:
                     yoe_param = {'range': {'jobs.experience': {'gte': 
yoe[0], 'lte': yoe[1]}}}
                 elif int(yoe[0]):
                     yoe_param = {'range': {'jobs.experience': {'gte': 
yoe[0]}}}
                 else:
                     yoe_param = {'term': {'jobs.experience': yoe[0]}}
             query_bool = {'must': [{'range': {'jobs.deadline': {'gt': 
str(date.today() + timedelta(days=1))}}}]}
             query_bool['must_not'] = [{'term': {'jobs.job_level': 
JOB_LEVEL_VOC}}]
             if specialization:
                 query_bool['must'].append(spec_param)
             if yoe:
                 query_bool['must'].append(yoe_param)
             es = config.get_es_connection()
             es_config = config.config['elasticsearch']
             # print({'query': {'bool': query_bool}})
             try:
                 # Tasks sometimes hang here
                 result = es.search(index=es_config['job_index'], 
doc_type=es_config['job_type'],
                                     body={'query': {'bool': query_bool}})
                 jobs = []
                 for j in result['hits']['hits']:
                     jobs.append(j['_source'])
             except ElasticsearchException as esc:
                 print(esc)
                 jobs = []

             if jobs:
                 memcache_client.set(key, jobs, 3600)
     except Exception as e:
         jobs = []
         print(e)

     return jobs

I find that the celery worker often stops executing tasks. After tests 
and debugging I in that this NEVER happens when I take out this line(s):
result = es.search(index=es_config['job_index'], 
doc_type=es_config['job_type'],
                                     body={'query': {'bool': query_bool}})

This line also does not raise any Exceptions

Does anyone have any idea what could be going on or how I can further 
inspect running tasks.
N.B celery worker is started with loglevel=debug flag but does not 
output any useful info as regards the problem.

Thanks