쿠...sal: [컴][웹][파이썬] celery beat 에서 schedule task 를 dynamically 추가/삭제

셀러리 비트 / celery beat how to add/remove scheduled task dynamically / 원하는 순간에 추가/삭제 / 다이내믹하게 /

celery beat 에서 schedule task 를 dynamically add/remove

git hub issue comment 에서 dynamic 하게 event 를 추가할 때 celery beat 가 새롭게 populate_heap() 을 하도록 하는 법이 나와 있다. 여기서 이야기하는 바는 아래 2가지이다.

populate_heap() 을 schedule 이 변경될 때마다 해주는 것
그리고 DatabaseSchduler 를 사용해서 add/remove 를 하는 것(참고: Using custom scheduler classes / Celery 4.1.0 documentation)

이중에 1번째 내용은 이슈의 comment에 의하면 4.1 버전에 반영되었다고 한다.

실제로 확인을 해보니 반영이 되어 있다. Populate heap when periodic tasks are changed · celery/celery@6a24e00 · GitHub

Scheduler

celery.beat.PersistentScheduler

기본적으로 celery.beat.PersistentScheduler를 사용한다.

$celery -A proj beat -s /home/celery/var/run/celerybeat-schedule

celery.beat.PersistentScheduler 는 celerybeat-schedule file을 이용하는데 shelve database 파일이다. 아래 shelve database 의 '제한사항(retrictions)' 에도 나오지만 concurrent access 를 지원하지 않는다.

https://docs.python.org/dev/library/shelve.html#restrictions

django-celery-beat

django-celery-beat 은 database 에 "periodic task schedule" 을 저장할 수 있도록 해준다. 하지만 이것은 Django 를 필요로 한다.

django-celery-beat - Database-backed Periodic Tasks — django_celery_beat 1.1.0 documentation

Django 없이 SQLAlchemy 를 이용하는 DatabaseScheduler

GitHub - tuomur/celery_sqlalchemy_scheduler: Celery scheduler that uses SQLAlchemy for storage.

사용법은 간단한다. SQLAlchemy 를 설치하고

$pip install sqlalchemy

아래 code 를 실행해서 DB table 을 생성한다.(참고로, 아래코드는 sqlite 을 사용하는 코드다)(다른 DB urls 과 관련해서는 여기 를 참고하자.)

참고로 추후에 sqlalchemy_scheduler_models.py 의 engine 도 user/password 를 해줘야 한다. 예를 들면 아래와 같은 식으로 말이다.

engine = sqlalchemy.create_engine('mysql://root:root_password@localhost/db_name')

## sqlalchemy_scheduler_models.py
...
engine = sqlalchemy.create_engine('sqlite://')
Base = declarative_base(bind=engine)
...

from sqlalchemy import create_engine
from sqlalchemy_scheduler_models import Base


# Create an engine that stores data in the local directory's
# sqlalchemy_example.db file.
engine = create_engine('sqlite:///sqlalchemy_example2.db')
 
# Create all tables in the engine. This is equivalent to "Create Table"
# statements in raw SQL.
Base.metadata.create_all(engine)

그리고 celconfig.py 에 아래를 추가하자.

CELERYBEAT_SCHEDULER = 'yourproject.sqlalchemy_scheduler:DatabaseScheduler'

주의

주의할 점은 scheduler 가 바뀌는 것에 대해서는 새롭게 sync 를 하지만 celery_crontabs 내용이 바뀌는 것에 대해서는 반영되지 않는다. 이것이 왜 중요하냐면, scheduler 에서 시간 설정을 crontab_id 로 하기 때문이다.

즉, crontab 의 내용을 변경하는 경우(시간을 변경하는 것)라면 새롭게 celery beat 를 restart 해야 한다.

etc

수행조건

enabled, last_run_at

table 'celery_schedules' 에 있는 task 들은 해당되는 crontab_id 시간에 수행된다. 이때 "enabled" 가 '1' 이 되어 있어야 하고, 또 하나 "last_run_at" 이 현재 UTC 보다 이전이어야 한다.

참고로 last_run_at 이라고 가정하고, 동작한다. 그러기 때문에 insert 할 때도 UTC 를 insert 해야 한다. 그렇지 않으면 last_run_at 이 현재 UTC 보다 큰 값이 돼서 동작이 실행되지 않을 수 있다.

sync 시점

또는 date_changed 를 보고 이전의 date_changed 시간이후여야 한다. 그래야 task 를 resync 하게 된다.(참고 : sqlalchemy_scheduler.py: DatabaseScheduler.should_sync())

should_sync 가 발생빈도는 Scheduler 의 sync_every 변수를 수정하면 된다.

# beat.py
class Scheduler(object):
     ...
    #: How often to sync the schedule (3 minutes by default)
    sync_every = 3 * 60

그리고 또하나 주의할 점은 session 과 관련돼서, session 이 기본적으로 autoflush 를 하지 않기 때문에, 새롭게 update 된 내용이 session 을 통해 이루어지지 않았다면, DatabaseScheduler 가 새롭게 sync 하지 못한다. (예를 들면, DB client 에서 직접 수정한 경우등)

그렇기 때문에 되도록 schedule 을 추가하는 부분도 동일한 session 을 이용하도록 하는 것이 낫다.

쿠...sal

[컴][웹][파이썬] celery beat 에서 schedule task 를 dynamically 추가/삭제