zengqingbo / mroget Goto Github PK

   目前发现部分任务在执行的时候，一直没有执行结果，导致容器无法处理任务。

   不知道如何监控hold的任务？

`app = Celery('tasks',
broker='redis://10.25.120.6/1',
backend='redis://10.25.120.6/1')

app.conf.update(
result_expires=3600,
worker_concurrency = 1,
)

logger = get_task_logger(name)

@app.task(queue='main')
def handleBigZipfile(filename,nid):
current_process()._config['daemon'] = False
logger.info('{} begin handle!'.format(filename))
filelist=handleAll(filename,nid)
if filelist != None : maininserttodb.delay(filelist,nid)
logger.info('{} is done!'.format(filename))

@app.task(queue='db')
def maintodb(filelist,nid):
conn = psycopg2.connect(PG_CONN_TEXT)
cur = conn.cursor()
for filename in filelist:
if filetype(filename) == 'cmpcounter' :
cur.execute("copy cmpcounter from '{}' WITH DELIMITER AS ',' NULL AS 'NIL' CSV;".format(filename) )
conn.commit()
cur.close()
conn.close()
os.remove(filename)

@app.task(queue='db')
def maininserttodb(filelist,nid):
conn = psycopg2.connect(PG_CONN_TEXT)
cur = conn.cursor()
for filename in filelist:
if filetype(filename) == 'cmpcounter' :
with open(filename) as csvfile:
for line in csvfile.readlines():
values = line.replace('NIL','null').replace(''','''').replace('"',''')
cur.execute('insert into cmpcounter values({});'.format(values))
conn.commit()
cur.close()
conn.close()
os.remove(filename)

def filetype(filename):
return os.path.splitext(os.path.splitext(filename)[0])[1].replace('.','').replace('.','')`

MRO大包处理过程中出现内存超限，导致docker服务被系统KILL的问题

celery任务处理异常导致 worker 退出

Traceback (most recent call last):
File "/usr/local/lib/python3.5/site-packages/celery/app/trace.py", line 374, in trace_task
R = retval = fun(*args, **kwargs)
File "/usr/local/lib/python3.5/site-packages/celery/app/trace.py", line 629, in protected_call
return self.run(*args, **kwargs)
File "/app/tasks.py", line 36, in handleBigZipfile
maintodb(csvfile)
File "/app/tasks.py", line 64, in maintodb
raise e
File "/app/tasks.py", line 59, in maintodb
cur.execute("copy {} from '{}' WITH DELIMITER AS ',' NULL AS 'NIL' CSV;".format(tablename,filename) )
psycopg2.DataError: no partition for partitioning key
CONTEXT: COPY nccounter, line 266: "2008-01-04 02:00:00.000,450201,115251458,NIL,NIL,NIL,1,NIL,NIL,NIL,NIL,NIL,2008-01-04 02:00:00.00"

若要限定容器在特定worker node運行，不要讓容器因停止而轉移到其他worker node，可以使用 constraint 參數，當然這可以寫成 docker-compose.yml ，若您不介意提供給我看貴公司所用的設定檔，我可以幫忙看一下。
https://docs.docker.com/engine/reference/commandline/service_create/#options