codemation / easyjobs Goto Github PK

first, GREAT(!) project and I believe it should deserve much more attention .

I pass through the documentation and the example but still have questions as i couldnt make it work as i expect,

so first for triggering a tasks flow without schedule it (by request)
i.e lets say i have basic flow :

             task2
          / 
task1 
         \
           taske 3  - task4

task2/task4 depends on task1
task3 depends on task3 and task2

worker:

server = FastAPI()

@server.on_event('startup')
async def setup():
    worker = await EasyJobsWorker.create(
        server,
        server_secret='abcd1234',
        manager_host='0.0.0.0',
        manager_port=8222,
        manager_secret='abcd1234',
        jobs_queue='ETL',
        max_tasks_per_worker=5
    )
    
    @worker.task()
    async def task1 (run_before=['task2', 'task3']):
        print(f"task1  - starting")
        await asyncio.sleep(5)
        print(f"task1 - finished")
        return f"task1!"
        
    @worker.task()
    async def task2(run_after=['task1']):
        print(f"task1 - starting")
        await asyncio.sleep(5)
        print(f"task2 - finished")
        return f"task2!"
        
    @worker.task(run_after=['task1'])
    async def task3(run_after=['task1']):
        print(f"task3 - starting")
        await asyncio.sleep(5)
        print(f"task3 - finished")
        return f"task3!"

    @worker.task()
    async def task4(run_after=['task3'],run_before=['task3']):
        print(f"task4 - starting")
        await asyncio.sleep(5)
        print(f"task4 - finished")
        return f"task4"

Based on the tasks plan i described above , should both run_after and run_before required?
As schedule is not set for none of the tasks ,
I expect that triggering task1 - will trigger the depended tasks automatically but it's not ,
It trigger just task1 .

04-30 20:31 EasyRpc-server /ws/jobs WARNING  worker 5fc6e54c_c8aa_11ec_bd8d_252c84f8bbc8 - pulled job {'job_id': '4d0c86ae-c8ab-11ec-bd8d-252c84f8bbc8', 'namespace': 'ETL', 'node_id': '5fc6e54c_c8aa_11ec_bd8d_252c84f8bbc8-REQ-60fcb1bc-c8aa-11ec-bd8d-252c84f8bbc8', 'status': 'reserved', 'name': 'task1', 'args': {'args': []}, 'kwargs': {'run_before': '[ "task2", "task3" ]'}, 'retry_policy': 'retry_once', 'on_failure': None, 'run_before': {'run_before': []}, 'run_after': {'run_after': []}, 'last_update': '2022-04-30T20:31:00'}
04-30 20:31 EasyRpc-server /ws/jobs WARNING  WORKER_MONITOR: working 1 / 5
task1  - starting
task1 - finished

Question regarding the pipline tasks:
is it possible to pipeline data between tasks - in example , the return values of task1 will use in task2?
Is there an option to reuse task in different nodes ? while providing kwargs dynamically so in I can trigger same task with different run_before and run_after ?

Suggest adding discussion tab and if it possible adding example folder that could be really helpful .
great project!

Background:

Under certain circumstances it can be advantageous to prevent workers from pulling new jobs to run.

• high worker utilization, or too many jobs running
• draining a worker for maintenance

Solution

Workers should have an method that can pause workers from consuming new work.

• Implement a pause workers method on EasyJobsServer
• Implement a pause workers method on EasyJobsWorker
• Create documentation for pause method usage & example