¿Cómo verificar el estado de la tarea en apio?

¿Cómo se comprueba si una tarea se está ejecutando en apio (específicamente, estoy usando celery-django)?

Leí la documentación y busqué en Google, pero no puedo ver una llamada como:

my_example_task.state() == RUNNING

Mi caso de uso es que tengo un servicio externo (java) para la transcodificación. Cuando envío un documento para que sea transcodificado, quiero verificar si la tarea que ejecuta ese servicio se está ejecutando y, de no ser así, (re) iniciarla.

Estoy usando las versiones estables actuales - 2.4, creo.

Devuelve el task_id (que se proporciona desde .delay ()) y luego pregunta a la instancia de apio sobre el estado:

x = method.delay(1,2)
print x.task_id

Cuando pregunte, obtenga un nuevo AsyncResult usando este task_id:

from celery.result import AsyncResult
res = AsyncResult("your-task-id")
res.ready()

La creación de un AsyncResultobjeto a partir de la identificación de la tarea es la forma recomendada en las Preguntas frecuentes para obtener el estado de la tarea cuando lo único que tiene es la identificación de la tarea.

Sin embargo, a partir de Celery 3.x, existen advertencias importantes que podrían morder a las personas si no les prestan atención. Realmente depende del escenario de caso de uso específico.

De forma predeterminada, Apio no registra un estado "en ejecución".

Para que Celery registre que se está ejecutando una tarea, debe establecer task_track_starteden True. Aquí hay una tarea simple que prueba esto:

@app.task(bind=True)
def test(self):
    print self.AsyncResult(self.request.id).state

Cuando task_track_startedes False, que es el valor predeterminado, el estado muestra PENDINGaunque la tarea ha comenzado. Si lo establece task_track_starteden True, entonces el estado será STARTED.

El estado PENDINGsignifica "No sé".

Una AsyncResultcon el estado PENDINGno significa nada más que que Apio desconoce el estado de la tarea. Esto podría deberse a varias razones.

Por un lado, AsyncResultse puede construir con identificadores de tareas no válidos. Dichas "tareas" serán consideradas pendientes por Celery:

>>> task.AsyncResult("invalid").status
'PENDING'

De acuerdo, nadie va a proporcionar identificadores obviamente inválidos AsyncResult. Bastante justo, pero también tiene como efecto que AsyncResulttambién considerará una tarea que se ha ejecutado con éxito pero que Celery ha olvidado como PENDING. Nuevamente, en algunos escenarios de casos de uso esto puede ser un problema. Parte del problema depende de cómo se configura Celery para mantener los resultados de las tareas, porque depende de la disponibilidad de las "lápidas" en el backend de resultados. ("Tombstones" es el término que se usa en la documentación de Celery para los fragmentos de datos que registran cómo terminó la tarea). El uso AsyncResultno funcionará en absoluto si task_ignore_resultes así True. Un problema más irritante es que Celery expira las lápidas por defecto. losresult_expiresLa configuración predeterminada está establecida en 24 horas. Entonces, si inicia una tarea y registra la identificación en el almacenamiento a largo plazo, y más 24 horas después, crea una AsyncResultcon ella, el estado será PENDING.

Todas las "tareas reales" comienzan en el PENDINGestado. Por lo tanto, realizar PENDINGuna tarea podría significar que se solicitó la tarea, pero nunca avanzó más allá de esto (por cualquier motivo). O podría significar que la tarea se ejecutó pero Celery olvidó su estado.

¡Ay! AsyncResultno funcionará para mí. ¿Que más puedo hacer?

Prefiero realizar un seguimiento de los objetivos que realizar un seguimiento de las tareas en sí . Conservo cierta información sobre las tareas, pero en realidad es secundaria al seguimiento de los objetivos. Las porterías se almacenan en almacenamiento independiente del apio. Cuando una solicitud necesita realizar un cálculo depende de que se haya logrado algún objetivo, verifica si el objetivo ya se ha logrado, en caso afirmativo, utiliza este objetivo almacenado en caché, de lo contrario, inicia la tarea que afectará el objetivo y envía a el cliente que hizo la solicitud HTTP una respuesta que indica que debe esperar un resultado.

Los nombres de variables y los hipervínculos anteriores son para Celery 4.x. En 3.x las variables y los enlaces correspondientes son los siguientes: CELERY_TRACK_STARTED, CELERY_IGNORE_RESULT, CELERY_TASK_RESULT_EXPIRES.

Cada Taskobjeto tiene una .requestpropiedad que lo contiene AsyncRequest. En consecuencia, la siguiente línea muestra el estado de una tarea task:

task.AsyncResult(task.request.id).state

También puede crear estados personalizados y actualizar su valor según la ejecución de la tarea. Este ejemplo es de docs:

@app.task(bind=True)
def upload_files(self, filenames):
    for i, file in enumerate(filenames):
        if not self.request.called_directly:
            self.update_state(state='PROGRESS',
                meta={'current': i, 'total': len(filenames)})

http://celery.readthedocs.org/en/latest/userguide/tasks.html#custom-states

Antigua pregunta, pero recientemente me encontré con este problema.

Si está intentando obtener el task_id, puede hacerlo así:

import celery
from celery_app import add
from celery import uuid

task_id = uuid()
result = add.apply_async((2, 2), task_id=task_id)

Ahora sabe exactamente qué es task_id y ahora puede usarlo para obtener AsyncResult:

# grab the AsyncResult 
result = celery.result.AsyncResult(task_id)

# print the task id
print result.task_id
09dad9cf-c9fa-4aee-933f-ff54dae39bdf

# print the AsyncResult's status
print result.status
SUCCESS

# print the result returned 
print result.result
4

Simplemente use esta API de las preguntas frecuentes sobre apio

result = app.AsyncResult(task_id)

Esto funciona bien.

Respuesta de 2020:

#### tasks.py
@celery.task()
def mytask(arg1):
    print(arg1)

#### blueprint.py
@bp.route("/args/arg1=<arg1>")
def sleeper(arg1):
    process = mytask.apply_async(args=(arg1,)) #mytask.delay(arg1)
    state = process.state
    return f"Thanks for your patience, your job {process.task_id} \
             is being processed. Status {state}"

Tratar:

task.AsyncResult(task.request.id).state

esto proporcionará el estado de la tarea de apio. Si la tarea de apio ya está en estado FALLO , arrojará una excepción:

raised unexpected: KeyError('exc_type',)

para tareas simples, podemos usar http://flower.readthedocs.io/en/latest/screenshots.html y http://policystat.github.io/jobtastic/ para hacer el monitoreo.

y para tareas complicadas, digamos una tarea que se ocupa de muchos otros módulos. Recomendamos registrar manualmente el progreso y el mensaje en la unidad de tarea específica.

Encontré información útil en el

Guía de trabajadores del proyecto de apio inspectores de trabajadores

En mi caso, estoy comprobando si Celery se está ejecutando.

inspect_workers = task.app.control.inspect()
if inspect_workers.registered() is None:
    state = 'FAILURE'
else:
    state = str(task.state) 

Puede jugar con inspeccionar para satisfacer sus necesidades.

• Primero, en su APLICACIÓN de apio:

vi my_celery_apps / app1.py

app = Celery(worker_name)

• y luego, cambie al archivo de tareas, importe la aplicación desde su módulo de aplicación de apio.

vi tasks / task1.py

from my_celery_apps.app1 import app

app.AsyncResult(taskid)

try:
   if task.state.lower() != "success":
        return
except:
    """ do something """

Aparte del enfoque programático anterior, se puede ver fácilmente el estado de la tarea de flor.

Monitoreo en tiempo real usando Celery Events. Flower es una herramienta basada en web para monitorear y administrar clústeres de apio.

1. Progreso e historial de tareas
2. Capacidad para mostrar detalles de la tarea (argumentos, hora de inicio, tiempo de ejecución y más)
3. Gráficos y estadísticas

Documento oficial: Flor - Herramienta de seguimiento del apio

Instalación:

$ pip install flower

Uso:

http://localhost:5555

res = method.delay()
    
print(f"id={res.id}, state={res.state}, status={res.status} ")

print(res.get())