録画後に１つずつエンコードさせるためにRabbitMQとCelery

ラズパイのハードウェアエンコードはパワーがないため １つずつエンコードさせるようにした

ジョブキューで管理できるようにbrokerにRabbitMQを採用し 見やすく管理しやすいようにworkerにCeleryも導入

もしラズパイだけじゃエンコード間に合わないのであれば 別マシンでキューを拾ってエンコードするということもできる

CeleryがpythonなのでWindowsでも行けるんじゃないかな？ まだやれてないから動くかわからんけど

RabbitMQ

パッケージインストール あとプラグインインストール

sudo apt-get install rabbitmq-server
sudo rabbitmq-plugins enable rabbitmq_management

サーバ起動

sudo service rabbitmq-server restart

chinachuのvhostとユーザを作成 権限付与したりなんたり

sudo rabbitmqctl add_vhost /chinachu
sudo rabbitmqctl add_user chinachu chinachu
sudo rabbitmqctl set_permissions -p /chinachu chinachu ".*" ".*" ".*"
sudo rabbitmqctl set_user_tags chinachu administrator

チェック

sudo rabbitmqctl list_users
sudo rabbitmqctl list_vhosts
sudo rabbitmqctl list_permissions -p /chinachu

ゲストユーザは削除

sudo rabbitmqctl delete_user guest

Celery

Celeryとflowerをインストールする flowerはCeleryのTask管理ツールで ブラウザからタスクの状態見れるので便利

sudo apt-get install python-pip
sudo pip install celery
sudo pip install flower

実際にキューの処理を行うworkerの作成 どっかで失敗したらexceptionを発生させる そうするとflowerブラウザ上で見えるから確認しやすいんで。 ffmpegのオプションは色々試して私はコレがよかった気がする

vim encode_task.py
------------------------------------------
# -*- coding: utf-8 -*-
import os
import subprocess
from celery import Celery

cmd = "ffmpeg -i -vcodec h264_omx -r 24000/1001 -vf yadif=0:-1:1,decimate -b:v 4000k -aspect 16:9 -s 1280x720 -acodec aac"
dst_ext = '.mp4'

app = Celery()
app.config_from_object('encode_task_config')

@app.task
def start(path):
    dst_path, ext = os.path.splitext(path)
    cmd_list = cmd.split()
    cmd_list.insert(2, path)
    cmd_list.append(dst_path + dst_ext)
    p = subprocess.Popen(cmd_list, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    out, err = p.communicate()
    if p.returncode == 0:
        os.remove(path)
        return 'Created :' + dst_path + dst_ext
    else:
        raise Exception("ReturnCode: " + str(p.returncode) + " [" + repr(err) + "]")

設定ファイル workerの同時処理数はコメントアウトしてデフォルトに 今はまだworkerは１つしかいないのでプリフェッチは0を設定して制限なしにする そんな録画量もないしね

vim encode_task_config.py
------------------------------------------
# message broker
BROKER_HOST = "localhost"
BROKER_PORT = 5672
BROKER_USER = "chinachu"
BROKER_PASSWORD = "chinachu"
BROKER_VHOST = "/chinachu"

# backend
CELERY_RESULT_BACKEND = "amqp"

# task
CELERY_ACKS_LATE = True
CELERYD_PREFETCH_MULTIPLIER = 0

# worker
#CELERYD_CONCURRENCY

# time zone
CELERY_ENABLE_UTC = False
CELERY_TIMEZONE = "Asia/Tokyo"

# log
#CELERYD_LOG_FILE = "celeryd.log"
# log level
CELERYD_LOG_LEVEL = "ERROR" # DEBUG, INFO, WARNING, ERROR or CRITICAL

# import module
CELERY_IMPORTS = ("encode_task", )

タスクを送るほうを作成 このpythonをChinachuの"recordedCommand"に設定したりする (実際はbash経由にしてました)

vim encode.py
--------------------------
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import sys
from encode_task import start

args = sys.argv
argc = len(args)

if(argc != 2):
    print 'Usage: $ python %s filepath' % args[0]
    quit()

start.delay(args[1])

Celeryサービス

Celeryサービスを作成する 下記のページにサンプル？があったので流用させてもらいます

celery/celery.service at master · celery/celery · GitHub github.com

sudo vim /etc/systemd/system/celery.service
--------------------------------------------------
[Unit]
Description=Celery Service
After=network.target

[Service]
Type=forking
User=chinachu
Group=chinachu
EnvironmentFile=-/etc/celery.conf
WorkingDirectory=/home/celery
ExecStart=/bin/sh -c '${CELERY_BIN} multi start $CELERYD_NODES \
        -A $CELERY_APP --pidfile=${CELERYD_PID_FILE} --logfile=${CELERYD_LOG_FILE} \
        --loglevel="${CELERYD_LOG_LEVEL}" $CELERYD_OPTS'
ExecStop=/bin/sh -c '${CELERY_BIN} multi stopwait $CELERYD_NODES \
        --pidfile=${CELERYD_PID_FILE}'
ExecReload=/bin/sh -c '${CELERY_BIN} multi restart $CELERYD_NODES \
        -A $CELERY_APP --pidfile=${CELERYD_PID_FILE} --logfile=${CELERYD_LOG_FILE} \
        --loglevel="${CELERYD_LOG_LEVEL}" $CELERYD_OPTS'

[Install]
WantedBy=multi-user.target

confファイル

sudo vim /etc/celery.conf
--------------------------------------------------
CELERY_APP="encode_task"
CELERYD_NODES="worker"
CELERYD_OPTS=""
CELERY_BIN="/usr/local/bin/celery"
CELERYD_PID_FILE="/var/run/celery/%n.pid"
CELERYD_LOG_FILE="/var/log/celery/%n%I.log"
CELERYD_LOG_LEVEL="ERROR"

起動時にディレクトリを作成するよう設定

sudo vim /etc/tmpfiles.d/celery.conf
--------------------------------------------------
d /var/run/celery 0755 chinachu chinachu -
d /var/log/celery 0755 chinachu chinachu -
--------------------------------------------------

flowerも同様にサービス化 (たしか途中でめんどくさくなって汎用的に作るのやめた)

sudo vim /etc/systemd/system/flower.service
--------------------------------------------------
[Unit]
Description=Flower Service
After=celery.service

[Service]
Type=simple
User=chinachu
Group=chinachu
PIDFile=/var/run/celery/flower.pid
WorkingDirectory=/home/chinachu/celery
ExecStart=/usr/local/bin/flower -A encode_task --logging=error
KillMode=process

[Install]
WantedBy=multi-user.target

サービス有効

sudo systemctl enable celery
sudo systemctl enable flower

以上！

こんな感じでブラウザからタスクの状態が見れる

---

ただエンコードするとChianchuの録画済ファイルパスのリンクが切れる REST APIを新しく追加し、エンコード後APIを叩きファイルパスを変更するようにした

気が向いたらまた書きます まだ下書きすらしてないので