from redis import Redis
from rq import Queue
from rq.job import Job
from rq.registry import FailedJobRegistry
redis = Redis()
queue = Queue(connection=redis)
registry = FailedJobRegistry(queue=queue)# Show all failed job IDs and the exceptions they caused during runtimefor job_id in registry.get_job_ids():
job = Job.fetch(job_id, connection=redis)print(job_id, job.exc_info)
08:39:14 default: say(msg='hello') (c97f39f8-c197-43ac-828f-245d78ef78f0)
Start say
{'msg': 'hello'}
08:42:14 [Job c97f39f8-c197-43ac-828f-245d78ef78f0]: exception raised while executing (say)
Traceback (most recent call last):
File "/Users/user/.local/share/virtualenvs/python-try-aR_k1rUJ/lib/python3.11/site-packages/rq/worker.py", line 1431, in perform_job
rv = job.perform()
^^^^^^^^^^^^^
File "/Users/user/.local/share/virtualenvs/python-try-aR_k1rUJ/lib/python3.11/site-packages/rq/job.py", line 1280, in perform
self._result = self._execute()
^^^^^^^^^^^^^^^
File "/Users/user/.local/share/virtualenvs/python-try-aR_k1rUJ/lib/python3.11/site-packages/rq/job.py", line 1317, in _execute
result = self.func(*self.args, **self.kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/user/data/repo/python-try/lib/util.py", line 8, in say
time.sleep(190)
File "/Users/user/.local/share/virtualenvs/python-try-aR_k1rUJ/lib/python3.11/site-packages/rq/timeouts.py", line 63, in handle_death_penalty
raise self._exception('Task exceeded maximum timeout value ({0} seconds)'.format(self._timeout))
rq.timeouts.JobTimeoutException: Task exceeded maximum timeout value (180 seconds)
08:52:14 default: say(msg='hello') (91aab157-25a5-44b9-898b-4da6dcb3baff)
Start say
{'msg': 'hello'}
End say
08:55:24 default: Job OK (91aab157-25a5-44b9-898b-4da6dcb3baff)
08:55:24 Result is kept for 500 seconds
11:21:53 default: say(msg='hello') (4d5a4db9-e8c9-4ee3-b64c-1dd266e66068)
Start say
11:21:55 Worker 98f9aac6f73445f685d4f5588abd514f [PID 32100]: warm shut down requested
{'msg': 'hello'}
11:22:03 default: Job OK (4d5a4db9-e8c9-4ee3-b64c-1dd266e66068)
11:22:03 Result is kept for 500 seconds
11:22:03 Worker rq:worker:98f9aac6f73445f685d4f5588abd514f: stopping on request
11:22:03 Unsubscribing from channel rq:pubsub:98f9aac6f73445f685d4f5588abd514f
kill -9 した場合
ワーカーがジョブを実行中に kill -9 を送信してみました
その場合ワーカーは Warm shut down せずにすぐに終了してしまいます
12:05:54 default: say(msg='hello') (b4eac96d-9591-482c-9753-e3a9c150d83b)
Start say
zsh: killed pipenv run rq worker
SIGTERM を受取り Warm shut down がスタートしますが10秒後に docker が強制的に SIGKILL を送信するためワーカーはジョブの完了を待たずに終了してしました
また強制終了したジョブは FailedJobRegistry に移動するようです
StartedJobRegistry cleanup: Moving job to FailedJobRegistry (due to AbandonedJobError)
time=60 にしてみる
SITTERM -> SIGKILL の送信にかかる時間を 60 秒に伸ばしてみます
docker stop --time=60 worker
すると今度はちゃんとワーカーがジョブの終了を待ってから停止していることが確認できました
03:32:54 default: say(msg='hello') (8e04f278-c5b1-4c4a-9069-3105960129f4)
03:33:14 Worker faa70ceee374410ba9416f55d74f4927 [PID 1]: warm shut down requested
Start say
{'msg': 'hello'}
03:33:25 default: Job OK (8e04f278-c5b1-4c4a-9069-3105960129f4)
03:33:25 Result is kept for 500 seconds
03:33:25 Worker rq:worker:faa70ceee374410ba9416f55d74f4927: stopping on request
03:33:25 Unsubscribing from channel rq:pubsub:faa70ceee374410ba9416f55d74f4927
from datetime import timedelta
from flask import Flask
from flask_rq2 import RQ
app = Flask(__name__)
app.config["RQ_REDIS_URL"]="redis://localhost:6379/0"
rq = RQ(app)classUser:def__init__(self, name)->None:
self.name = name
defsay(self, msg):print(msg)
@app.route("/register")defregister_job():
scheduler = rq.get_scheduler(queue="bar")
job = scheduler.enqueue_in(timedelta(seconds=5), User("hawksnowlog").say,"HELLO")return job.get_id()
動作確認
まずはスケジューラを起動します
pipenv run rqscheduler
13:40:50 Registering birth
ここの時間が起点時間になります
次にワーカーを起動します
pipenv run rq worker bar
あとはアプリを起動しジョブを登録します
pipenv run flask run
curl localhost:5000/register
するとワーカーのジョブに起点時間から次の 1 分後にログが流れることが確認できると思います
13:41:50 bar: say('HELLO') (0378cd1a-274f-46f7-8484-00d513b98413)
HELLO
13:41:50 bar: Job OK (0378cd1a-274f-46f7-8484-00d513b98413)
13:41:50 Result is kept for 500 seconds
ジョブ登録時に meta 引数を指定することができます
辞書で好きな値を指定できます
pickle するので pickle 可能なオブジェクトにしましょう
環境
Python 3.11.3
flask-rq2 18.3
rq-scheduler 0.13.1
サンプルコード
from flask import Flask, request
from flask_rq2 import RQ
app = Flask(__name__)
app.config["RQ_REDIS_URL"]="redis://localhost:6379/0"
rq = RQ(app)defhello():raise Exception()
@app.route("/")defget_jobs():# 検索する id をクエリストリングで渡せるid= request.args.get("id")
scheduler = rq.get_scheduler(queue="bar")
jobs_ = scheduler.get_jobs()# id が設定されていない場合は全件取得、そうでない場合は meta 情報を使って id 検索ifidisNone:
jobs = jobs_
else:
jobs =[
j for j in jobs_ if j.meta.get("id")isnotNoneand j.meta.get("id")==id]return[{"id": job.id,"meta": job.meta.get("id"),}for job in jobs
]
@app.route("/register")defregister_job():# id を meta 情報として設定するid= request.args.get("id")
scheduler = rq.get_scheduler(queue="bar")
job = scheduler.cron("*/1 * * * *",
func=hello,
args=[],
meta={"id":id},# meta 属性に辞書を設定可能)return job.get_id()
最後に
ジョブを特定する際に使えると思います
ただ meta 情報で直接検索する機能がないので全件取得してから検索するしかないので redis のデータが膨大になる場合にはパフォーマンスなどの注意が必要です
15:50:00 Worker rq:worker:9f0f761294194e79a4bb9b87c0e35436 started with PID 90351, version 1.15.0
15:50:00 Subscribing to channel rq:pubsub:9f0f761294194e79a4bb9b87c0e35436
15:50:00 *** Listening on bar...
15:50:00 Cleaning registries for queue: bar
15:50:00 bar: lib.util.say('hello') (53acc2fb-aa88-4627-9a4c-183eab6ee07a)
hello
15:50:00 bar: Job OK (53acc2fb-aa88-4627-9a4c-183eab6ee07a)
15:50:00 Result will never expire, clean up result key manually
