Running Apscheduler in multiple instances

[PraveenKumarM21]

@agronholm I'm trying to deploy app scheduler in my flask application which schedules task via Api calls. I have initiated scheduler in the background. This application runs as a Kubernetes pod in a cluster. Everything was fine until the application is deployed as three instances so if one goes down the other one can take up its place. This causes duplication of tasks.

Is there any way to resolve this so that if one scheduler instance runs the schedules the other may not interrupt?

I'm using Apscheduler 4.0.0a5 with MongoDB as data_ store and redis as event broker.

[agronholm]

APScheduler uses data store atomicity and locking mechanisms to ensure that each job is only acquired by exactly one scheduler. Can you show me your code, as this should not be happening? I also advise you to use code from master instead of v4.0.0a5, as there are a lot of fixes that have not been published in the latest pre-release.

[PraveenKumarM21]

import pytz
from apscheduler import Scheduler, TaskDefaults
from apscheduler.triggers.date import DateTrigger
from apscheduler.datastores.mongodb import MongoDBDataStore
from apscheduler.eventbrokers.redis import RedisEventBroker
from apscheduler.executors.thread import ThreadPoolJobExecutor
from datetime import datetime
from flask import Flask, request, jsonify
from uuid import uuid4

def run_task():
print(f"Task Executed Successfully
{datetime.now()}")

mongo_uri= 'mongodb://localhost:27017'
redis_uri= 'redis://localhost:6379/0'
database = 'scheduler'

data_store = MongoDBDataStore(client_or_uri=mongo_uri, database=database)
sched_config = dict(data_store=data_store, job_executors={'threadpool': ThreadPoolJobExecutor(max_workers=1)},
task_defaults=TaskDefaults( job_executor="threadpool"),)
event = RedisEventBroker(client_or_url=redis_uri)
sched = Scheduler(**sched_config, event_broker=event_broker)
app = Flask(name)

@app.route('/trigger', methods=['POST'])
def schedule_task():

data = request.json
date_str = data["date"]
hour = int(data["hour"])
minute = int(data["minute"])
timezone_str = data["timezone"]
local timezone = pytz.timezone(timezone_str)
local_datetime = local timezone.localize(
datetime(year=int(data_str.split("-")[0]), month
int(date_str.split("-")[1], day=int(date_str.split("-
")[2]), hour=hour, minute=minute))
utc_datetime =
local_datetime.astimezone(pytz.utc)
sched.add_schedule(run_task,
trigger=DateTrigger(run_time=utc_datetime),
id=str(uuid4()))
result = f'Task scheduled at {data["hour"]}
{data["minute"]} {data["date"]}'
print(result)
return jsonify({"result": result})

if name == 'main':
sched.start_in_background()
app.run(host='0.0.0.0', port=8080,
use_reloader=False)

[PraveenKumarM21]

So this will occur even if I use postgresql???

[agronholm]

No, the bug I found was in the MongoDB data store. On the SQLAlchemy store I use transactions and pessimistic locking which will prevent this altogether.

[agronholm]

I will fix this in the MongoDB data store over the weekend.

[agronholm]

Fixed via 4758e8f.

[PraveenKumarM21]

Thanks

[PraveenKumarM21]

@agronholm can you accommodate DocumentDB like how you did with MongoDB and SQLAlchemy for datastores

[agronholm]

You asked me that before, and my answer is still the same: I don't know how. If you can find a trick in pymongo that provides compatibility, I can try, but I won't go out of my way to do that.