mirror of
https://github.com/joshuaboniface/rffmpeg.git
synced 2026-07-18 00:45:53 +00:00
Merge pull request #9 from crisidev/master
This commit is contained in:
Joshua M. Boniface
GitHub
commit
d0fd57309f
3 changed files with 138 additions and 121 deletions
|
|
@ -24,6 +24,9 @@ rffmpeg is a remote FFmpeg wrapper used to execute FFmpeg commands on a remote s
|
|||
|
||||
rffmpeg supports setting multiple hosts. It keeps state in `/run/shm/rffmpeg`, of all running processes. These state files are used during rffmpeg's initialization in order to determine the optimal target host. rffmpeg will run through these hosts sequentially, choosing the one with the fewest running rffmpeg jobs. This helps distribute the transcoding load across multiple servers, and can also provide redundancy if one of the servers is offline - rffmpeg will detect if a host is unreachable and set it "bad" for the remainder of the run, thus skipping it until the process completes.
|
||||
|
||||
### Local host
|
||||
If one of the hosts in the config file is called "localhost", rffmpeg will run locally without SSH. This can be usefull if you have a host which serves the frontend (IE Jellyfin) and is also able to transcode.
|
||||
|
||||
### Terminating rffmpeg
|
||||
|
||||
When running rffmpeg manually, *do not* exit it with `Ctrl+C`. Doing so will likely leave the `ffmpeg` process running on the remote machine. Instead, enter `q` and a newline ("Enter") into the rffmpeg process, and this will terminate the entire command cleanly. This is the method that Jellyfin uses to communicate the termination of an `ffmpeg` process.
|
||||
|
|
|
|||
254
rffmpeg.py
254
rffmpeg.py
|
|
@ -36,71 +36,67 @@
|
|||
# Imports and helper functions
|
||||
###############################################################################
|
||||
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import re
|
||||
import yaml
|
||||
import subprocess
|
||||
import signal
|
||||
from datetime import datetime
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
def logger(msg):
|
||||
log_to_file = config.get('log_to_file', False)
|
||||
logfile = config.get('logfile', False)
|
||||
import yaml
|
||||
|
||||
log = logging.getLogger("rffmpeg")
|
||||
|
||||
if log_to_file and logfile:
|
||||
with open(logfile, 'a') as logfhd:
|
||||
logfhd.write(str(datetime.now()) + ' ' + str(msg) + '\n')
|
||||
|
||||
###############################################################################
|
||||
# Configuration parsing
|
||||
###############################################################################
|
||||
|
||||
# Get configuration file
|
||||
default_config_file = '/etc/rffmpeg/rffmpeg.yml'
|
||||
config_file = os.environ.get('RFFMPEG_CONFIG', default_config_file)
|
||||
default_config_file = "/etc/rffmpeg/rffmpeg.yml"
|
||||
config_file = os.environ.get("RFFMPEG_CONFIG", default_config_file)
|
||||
|
||||
# Parse the configuration
|
||||
with open(config_file, 'r') as cfgfile:
|
||||
with open(config_file, "r") as cfgfile:
|
||||
try:
|
||||
o_config = yaml.load(cfgfile, Loader=yaml.BaseLoader)
|
||||
except Exception as e:
|
||||
logger('ERROR: Failed to parse configuration file: {}'.format(e))
|
||||
log.error("ERROR: Failed to parse configuration file: %s", e)
|
||||
exit(1)
|
||||
|
||||
try:
|
||||
config = {
|
||||
'state_tempdir': o_config['rffmpeg']['state']['tempdir'],
|
||||
'state_filename': o_config['rffmpeg']['state']['filename'],
|
||||
'state_contents': o_config['rffmpeg']['state']['contents'],
|
||||
'log_to_file': o_config['rffmpeg']['logging']['file'],
|
||||
'logfile': o_config['rffmpeg']['logging']['logfile'],
|
||||
'remote_hosts': o_config['rffmpeg']['remote']['hosts'],
|
||||
'remote_user': o_config['rffmpeg']['remote']['user'],
|
||||
'remote_args': o_config['rffmpeg']['remote']['args'],
|
||||
'pre_commands': o_config['rffmpeg']['commands']['pre'],
|
||||
'ffmpeg_command': o_config['rffmpeg']['commands']['ffmpeg'],
|
||||
'ffprobe_command': o_config['rffmpeg']['commands']['ffprobe']
|
||||
"state_tempdir": o_config["rffmpeg"]["state"]["tempdir"],
|
||||
"state_filename": o_config["rffmpeg"]["state"]["filename"],
|
||||
"state_contents": o_config["rffmpeg"]["state"]["contents"],
|
||||
"log_to_file": o_config["rffmpeg"]["logging"]["file"],
|
||||
"logfile": o_config["rffmpeg"]["logging"]["logfile"],
|
||||
"remote_hosts": o_config["rffmpeg"]["remote"]["hosts"],
|
||||
"remote_user": o_config["rffmpeg"]["remote"]["user"],
|
||||
"remote_args": o_config["rffmpeg"]["remote"]["args"],
|
||||
"pre_commands": o_config["rffmpeg"]["commands"]["pre"],
|
||||
"ffmpeg_command": o_config["rffmpeg"]["commands"]["ffmpeg"],
|
||||
"ffprobe_command": o_config["rffmpeg"]["commands"]["ffprobe"],
|
||||
}
|
||||
except Exception as e:
|
||||
logger('ERROR: Failed to load configuration: {}'.format(e))
|
||||
log.error("ERROR: Failed to load configuration: %s", e)
|
||||
exit(1)
|
||||
|
||||
# Handle the fallback configuration using get() to avoid failing
|
||||
config['fallback_ffmpeg_command'] = o_config['rffmpeg']['commands'].get('fallback_ffmpeg', config['ffmpeg_command'])
|
||||
config['fallback_ffprobe_command'] = o_config['rffmpeg']['commands'].get('fallback_ffprobe', config['ffprobe_command'])
|
||||
config["fallback_ffmpeg_command"] = o_config["rffmpeg"]["commands"].get("fallback_ffmpeg", config["ffmpeg_command"])
|
||||
config["fallback_ffprobe_command"] = o_config["rffmpeg"]["commands"].get("fallback_ffprobe", config["ffprobe_command"])
|
||||
|
||||
# Parse CLI args (ffmpeg command line)
|
||||
all_args = sys.argv
|
||||
cli_ffmpeg_args = all_args[1:]
|
||||
|
||||
# Get PID
|
||||
our_pid = os.getpid()
|
||||
current_statefile = config['state_tempdir'] + '/' + config['state_filename'].format(pid=our_pid)
|
||||
current_statefile = config["state_tempdir"] + "/" + config["state_filename"].format(pid=os.getpid())
|
||||
|
||||
logger("Starting rffmpeg {}: {}".format(our_pid, ' '.join(all_args)))
|
||||
log.info("Starting rffmpeg %s: %s", os.getpid(), " ".join(all_args))
|
||||
|
||||
def local_ffmpeg_fallback():
|
||||
|
||||
def run_local_ffmpeg():
|
||||
"""
|
||||
Fallback call to local ffmpeg
|
||||
"""
|
||||
|
|
@ -112,66 +108,56 @@ def local_ffmpeg_fallback():
|
|||
stderr = sys.stderr
|
||||
|
||||
# Verify if we're in ffmpeg or ffprobe mode
|
||||
if 'ffprobe' in all_args[0]:
|
||||
rffmpeg_command.append(config['fallback_ffprobe_command'])
|
||||
if "ffprobe" in all_args[0]:
|
||||
rffmpeg_command.append(config["fallback_ffprobe_command"])
|
||||
stdout = sys.stdout
|
||||
else:
|
||||
rffmpeg_command.append(config['fallback_ffmpeg_command'])
|
||||
rffmpeg_command.append(config["fallback_ffmpeg_command"])
|
||||
|
||||
# Determine if version, encorders, or decoders is an argument; if so, we output stdout to stdout
|
||||
# Weird workaround for something Jellyfin requires...
|
||||
specials = ['-version', '-encoders', '-decoders', '-hwaccels']
|
||||
specials = ["-version", "-encoders", "-decoders", "-hwaccels"]
|
||||
if any(item in specials for item in cli_ffmpeg_args):
|
||||
stdout = sys.stdout
|
||||
|
||||
# Parse and re-quote any problematic arguments
|
||||
for arg in cli_ffmpeg_args:
|
||||
rffmpeg_command.append('{}'.format(arg))
|
||||
rffmpeg_command.append("{}".format(arg))
|
||||
|
||||
p = subprocess.run(rffmpeg_command,
|
||||
shell=False,
|
||||
bufsize=0,
|
||||
universal_newlines=True,
|
||||
stdin=stdin,
|
||||
stderr=stderr,
|
||||
stdout=stdout)
|
||||
returncode = p.returncode
|
||||
log.info("Local command for rffmpeg %s: %s", os.getpid(), " ".join(rffmpeg_command))
|
||||
|
||||
try:
|
||||
os.remove(current_statefile)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
returncode = run_command(rffmpeg_command, stdin, stdout, stderr)
|
||||
log.info("Finished local rffmpeg %s with return code %s", os.getpid(), returncode)
|
||||
return returncode
|
||||
|
||||
logger("Finished rffmpeg {} (local failover mode) with return code {}".format(our_pid, returncode))
|
||||
exit(returncode)
|
||||
|
||||
def get_target_host():
|
||||
"""
|
||||
Determine the optimal target host
|
||||
"""
|
||||
logger("Determining target host")
|
||||
log.info("Determining target host")
|
||||
|
||||
# Ensure the state directory exists or create it
|
||||
if not os.path.exists(config['state_tempdir']):
|
||||
os.makedirs(config['state_tempdir'])
|
||||
if not os.path.exists(config["state_tempdir"]):
|
||||
os.makedirs(config["state_tempdir"])
|
||||
|
||||
# Check for existing state files
|
||||
state_files = os.listdir(config['state_tempdir'])
|
||||
state_files = os.listdir(config["state_tempdir"])
|
||||
|
||||
# Read each statefile to determine which hosts are bad or in use
|
||||
bad_hosts = list()
|
||||
active_hosts = list()
|
||||
for state_file in state_files:
|
||||
with open(config['state_tempdir'] + '/' + state_file, 'r') as statefile:
|
||||
with open(config["state_tempdir"] + "/" + state_file, "r") as statefile:
|
||||
contents = statefile.readlines()
|
||||
for line in contents:
|
||||
if re.match('^badhost', line):
|
||||
if re.match("^badhost", line):
|
||||
bad_hosts.append(line.split()[1])
|
||||
else:
|
||||
active_hosts.append(line.split()[0])
|
||||
|
||||
# Get the remote hosts list from the config
|
||||
remote_hosts = config['remote_hosts']
|
||||
remote_hosts = config["remote_hosts"]
|
||||
|
||||
# Remove any bad hosts from the remote_hosts list
|
||||
for host in bad_hosts:
|
||||
|
|
@ -197,20 +183,22 @@ def get_target_host():
|
|||
target_host = host
|
||||
|
||||
# Write to our state file
|
||||
with open(current_statefile, 'a') as statefile:
|
||||
statefile.write(config['state_contents'].format(host=target_host) + '\n')
|
||||
with open(current_statefile, "a") as statefile:
|
||||
statefile.write(config["state_contents"].format(host=target_host) + "\n")
|
||||
|
||||
if not target_host:
|
||||
logger('Failed to find a valid target host - using local fallback instead')
|
||||
local_ffmpeg_fallback()
|
||||
log.error("Failed to find a valid target host - using local fallback instead")
|
||||
target_host = "localhost"
|
||||
|
||||
log.info("Selected valid target host %s", target_host)
|
||||
return target_host
|
||||
|
||||
|
||||
def bad_host(target_host):
|
||||
logger("Setting bad host {}".format(target_host))
|
||||
log.info("Setting bad host %s", target_host)
|
||||
|
||||
# Rewrite the statefile, removing all instances of the target_host that were added before
|
||||
with open(current_statefile, 'r+') as statefile:
|
||||
with open(current_statefile, "r+") as statefile:
|
||||
new_statefile = statefile.readlines()
|
||||
statefile.seek(0)
|
||||
for line in new_statefile:
|
||||
|
|
@ -221,37 +209,42 @@ def bad_host(target_host):
|
|||
# Add the bad host to the statefile
|
||||
# This will affect this run, as well as any runs that start while this one is active; once
|
||||
# this run is finished and its statefile removed, however, the host will be retried again
|
||||
with open(current_statefile, 'a') as statefile:
|
||||
statefile.write("badhost " + config['state_contents'].format(host=target_host) + '\n')
|
||||
with open(current_statefile, "a") as statefile:
|
||||
statefile.write("badhost " + config["state_contents"].format(host=target_host) + "\n")
|
||||
|
||||
|
||||
def setup_command(target_host):
|
||||
"""
|
||||
Craft the target command
|
||||
"""
|
||||
logger("Crafting remote command string")
|
||||
log.info("Crafting remote command string")
|
||||
|
||||
rffmpeg_command = list()
|
||||
|
||||
# Add SSH component
|
||||
rffmpeg_command.append('ssh')
|
||||
rffmpeg_command.append('-q')
|
||||
rffmpeg_command.append("ssh")
|
||||
rffmpeg_command.append("-q")
|
||||
|
||||
# Set our connection timeouts, in case one of several remote machines is offline
|
||||
rffmpeg_command.append('-o')
|
||||
rffmpeg_command.append('ConnectTimeout=1')
|
||||
rffmpeg_command.append('-o')
|
||||
rffmpeg_command.append('ConnectionAttempts=1')
|
||||
rffmpeg_command.append("-o")
|
||||
rffmpeg_command.append("ConnectTimeout=1")
|
||||
rffmpeg_command.append("-o")
|
||||
rffmpeg_command.append("ConnectionAttempts=1")
|
||||
rffmpeg_command.append("-o")
|
||||
rffmpeg_command.append("StrictHostKeyChecking=no")
|
||||
rffmpeg_command.append("-o")
|
||||
rffmpeg_command.append("UserKnownHostsFile=/dev/null")
|
||||
|
||||
for arg in config['remote_args']:
|
||||
for arg in config["remote_args"]:
|
||||
if arg:
|
||||
rffmpeg_command.append(arg)
|
||||
|
||||
# Add user+host string
|
||||
rffmpeg_command.append('{}@{}'.format(config['remote_user'], target_host))
|
||||
logger("Running rffmpeg {} on {}@{}".format(our_pid, config['remote_user'], target_host))
|
||||
rffmpeg_command.append("{}@{}".format(config["remote_user"], target_host))
|
||||
log.info("Running rffmpeg %s on %s@%s", os.getpid(), config["remote_user"], target_host)
|
||||
|
||||
# Add any pre command
|
||||
for cmd in config['pre_commands']:
|
||||
for cmd in config["pre_commands"]:
|
||||
if cmd:
|
||||
rffmpeg_command.append(cmd)
|
||||
|
||||
|
|
@ -261,82 +254,101 @@ def setup_command(target_host):
|
|||
stderr = sys.stderr
|
||||
|
||||
# Verify if we're in ffmpeg or ffprobe mode
|
||||
if 'ffprobe' in all_args[0]:
|
||||
rffmpeg_command.append(config['ffprobe_command'])
|
||||
if "ffprobe" in all_args[0]:
|
||||
rffmpeg_command.append(config["ffprobe_command"])
|
||||
stdout = sys.stdout
|
||||
else:
|
||||
rffmpeg_command.append(config['ffmpeg_command'])
|
||||
rffmpeg_command.append(config["ffmpeg_command"])
|
||||
|
||||
# Determine if version, encorders, or decoders is an argument; if so, we output stdout to stdout
|
||||
# Weird workaround for something Jellyfin requires...
|
||||
if '-version' in cli_ffmpeg_args or '-encoders' in cli_ffmpeg_args or '-decoders' in cli_ffmpeg_args:
|
||||
if "-version" in cli_ffmpeg_args or "-encoders" in cli_ffmpeg_args or "-decoders" in cli_ffmpeg_args:
|
||||
stdout = sys.stdout
|
||||
|
||||
# Parse and re-quote any problematic arguments
|
||||
for arg in cli_ffmpeg_args:
|
||||
# Match bad shell characters: * ( ) whitespace
|
||||
if re.search('[*()\s|\[\]]', arg):
|
||||
if re.search("[*()\s|\[\]]", arg):
|
||||
rffmpeg_command.append('"{}"'.format(arg))
|
||||
else:
|
||||
rffmpeg_command.append('{}'.format(arg))
|
||||
rffmpeg_command.append("{}".format(arg))
|
||||
|
||||
return rffmpeg_command, stdin, stdout, stderr
|
||||
|
||||
def prepare_command():
|
||||
logger("Preparing remote command")
|
||||
|
||||
target_host = get_target_host()
|
||||
rffmpeg_command, stdin, stdout, stderr = setup_command(target_host)
|
||||
rffmpeg_cli = ' '.join(rffmpeg_command)
|
||||
logger("Remote command for rffmpeg {}: {}".format(our_pid, rffmpeg_cli))
|
||||
|
||||
return rffmpeg_command, target_host, stdin, stdout, stderr
|
||||
|
||||
def run_command(rffmpeg_command, stdin, stdout, stderr):
|
||||
"""
|
||||
Execute the remote command using subprocess
|
||||
"""
|
||||
logger("Running remote command")
|
||||
log.info("Running command %s", " ".join(rffmpeg_command))
|
||||
|
||||
p = subprocess.run(rffmpeg_command,
|
||||
shell=False,
|
||||
bufsize=0,
|
||||
universal_newlines=True,
|
||||
stdin=stdin,
|
||||
stderr=stderr,
|
||||
stdout=stdout)
|
||||
p = subprocess.run(
|
||||
rffmpeg_command, shell=False, bufsize=0, universal_newlines=True, stdin=stdin, stderr=stderr, stdout=stdout
|
||||
)
|
||||
returncode = p.returncode
|
||||
|
||||
return returncode
|
||||
|
||||
def cleanup(signum='', frame=''):
|
||||
|
||||
def run_remote_ffmpeg(target_host):
|
||||
rffmpeg_command, stdin, stdout, stderr = setup_command(target_host)
|
||||
log.info("Remote command for rffmpeg %s: %s", os.getpid(), " ".join(rffmpeg_command))
|
||||
return run_command(rffmpeg_command, stdin, stdout, stderr)
|
||||
|
||||
|
||||
def cleanup(signum="", frame=""):
|
||||
# Remove the current statefile
|
||||
try:
|
||||
os.remove(current_statefile)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
signal.signal(signal.SIGTERM, cleanup)
|
||||
signal.signal(signal.SIGINT, cleanup)
|
||||
signal.signal(signal.SIGQUIT, cleanup)
|
||||
signal.signal(signal.SIGHUP, cleanup)
|
||||
|
||||
# Main process loop; executes until the ffmpeg command actually runs on a reachable host
|
||||
while True:
|
||||
logger("Starting process loop")
|
||||
def main():
|
||||
signal.signal(signal.SIGTERM, cleanup)
|
||||
signal.signal(signal.SIGINT, cleanup)
|
||||
signal.signal(signal.SIGQUIT, cleanup)
|
||||
signal.signal(signal.SIGHUP, cleanup)
|
||||
|
||||
# Set up and execute our command
|
||||
rffmpeg_command, target_host, stdin, stdout, stderr = prepare_command()
|
||||
returncode = run_command(rffmpeg_command, stdin, stdout, stderr)
|
||||
|
||||
# A returncode of 255 means that the SSH process failed; ffmpeg does not throw this return code (https://ffmpeg.org/pipermail/ffmpeg-user/2013-July/016245.html)
|
||||
if returncode == 255:
|
||||
logger("SSH failed to host {} with retcode {}: marking this host as bad and retrying".format(target_host, returncode))
|
||||
bad_host(target_host)
|
||||
log_to_file = config.get("log_to_file", False)
|
||||
if log_to_file:
|
||||
logfile = config.get("logfile")
|
||||
logging.basicConfig(
|
||||
filename=logfile, level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
||||
)
|
||||
else:
|
||||
# The SSH succeeded, so we can abort the loop
|
||||
break
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
|
||||
|
||||
cleanup()
|
||||
logger("Finished rffmpeg {} with return code {}".format(our_pid, returncode))
|
||||
exit(returncode)
|
||||
# Main process loop; executes until the ffmpeg command actually runs on a reachable host
|
||||
returncode = 1
|
||||
while True:
|
||||
log.info("Starting process loop")
|
||||
|
||||
target_host = get_target_host()
|
||||
print("[rffmpeg] Running on {}".format(target_host))
|
||||
if target_host == "localhost":
|
||||
returncode = run_local_ffmpeg()
|
||||
break
|
||||
else:
|
||||
returncode = run_remote_ffmpeg(target_host)
|
||||
|
||||
# A returncode of 255 means that the SSH process failed;
|
||||
# ffmpeg does not throw this return code (https://ffmpeg.org/pipermail/ffmpeg-user/2013-July/016245.html)
|
||||
if returncode == 255:
|
||||
log.info(
|
||||
"SSH failed to host %s with retcode %s: marking this host as bad and retrying",
|
||||
target_host,
|
||||
returncode,
|
||||
)
|
||||
bad_host(target_host)
|
||||
else:
|
||||
# The SSH succeeded, so we can abort the loop
|
||||
break
|
||||
|
||||
cleanup()
|
||||
log.info("Finished rffmpeg %s with return code %s", os.getpid(), returncode)
|
||||
exit(returncode)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
|
|||
|
|
@ -27,6 +27,7 @@ rffmpeg:
|
|||
remote:
|
||||
# A YAML list of remote hosts to connect to
|
||||
hosts:
|
||||
- localhost
|
||||
- gpu1
|
||||
# - gpu2
|
||||
# - gpu3
|
||||
|
|
@ -40,6 +41,7 @@ rffmpeg:
|
|||
- "-i"
|
||||
- "/var/lib/jellyfin/id_rsa"
|
||||
|
||||
|
||||
# Remote command configuration
|
||||
commands:
|
||||
# A YAML list of prefixes to the ffmpeg command (e.g. sudo, nice, etc.),
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue