Any method can get all flow jobs？

[lvscup]

Any method can get all flow jobs？

[fridex]

Question 1:
If I don't store all dispatcher ids after "run_flow", any method can list the dispatcher ids directly?

If you do not explicitly store these ids, there is no way how to retrieve them as both worker and flow/task producer are stateless. State is kept only on the queue (so by picking messages from the queue, you are able to obtain these ids or by explicitly storing them).

Question 2:
At one point, job may be in one of four status(Queued/Running/Finished/Failed). Tracing can monitor jobs that have started(Running/Finished/Failed), but any simple method list jobs that still stay in queue?
assume redis as broker.

Explicitly storing dispatcher ids on producer side is one option (and matching it against the tracing output). Another option is to use redis and retrieve states of tasks. This option is however too low level to me, if there will be any change in Celery, this solution can break.

I don't know your use case, but it sounds to me it could be a good idea in your system to introduce "monitoring" entity which would track jobs from producer side (a flow has been scheduled) and then create a tracing function which will notify the monitoring entity about the flow state (failure/runnig/finished/failed). If the solution would be generic enough, I'm happy to accept it in Selinon ecosystem :). It can be a simple REST API with some database for persistence. A UI can be then build to expose status of flows/tasks.

[lvscup]

Yes，I will track jobs from producer side as you said. If Selinon supports it later, I will be happy.

But another question arises: dispatcher_id can't assign before invoke "run_flow", if I use the dispatcher_id returned by run_flow method , then store it, the state of entity may be incorrect.
For example, t1: invoke run_flow, t2: event=FLOW_START(update state, but dispatcher_id hasn't stored yet). Because the call is asynchronous.

Celery apply_async method accepts the param 'task_id', before invoke apply_async method, init task_id as dispathcer_id, e.g. Dispatcher().apply_async(kwargs=, queue=,task_id=). if Selinon can support it, the problem of incorrect state will not exist.

[fridex]

For example, t1: invoke run_flow, t2: event=FLOW_START(update state, but dispatcher_id hasn't stored yet). Because the call is asynchronous.

Once you call run_flow, the flow will be scheduled (it hasn't been started yet) and once a worker picks dispatcher task (task responsible for scheduling the actual workload based on configuration), there will be emitted FLOW_START. Maybe it would worth it to add FLOW_SCHEDULED event to track also flow scheduling see https://selinon.readthedocs.io/en/latest/selinon/doc/selinon.trace.html#module-selinon.trace I'm open for PRs or I can do it if you want.

Anyway, if you call run_flow and there is no exception, you can expect the flow is scheduled for execution. I don't see how the task_id argument would help here.

[lvscup]

It will be great if Selinon has FLOW_SCHEDULED event. Use tracing function can track the life cycle of jobs. Then the issue about task_id can be ignored. Expect the new event. Thank you.

[fridex]

I'm closing this one, considering it as resolved. Feel free to reopen/post new issue in case of more questions.