From 8fba9d5a7f221524c9ff96534187e9802779cf8e Mon Sep 17 00:00:00 2001 From: Bigo Date: Sat, 6 Feb 2021 14:53:13 +0000 Subject: [PATCH 1/2] Modernize code and logging, simplify localhost handling --- README.md | 3 + rffmpeg.py | 254 ++++++++++++++++++++++++--------------------- rffmpeg.yml.sample | 4 + 3 files changed, 140 insertions(+), 121 deletions(-) diff --git a/README.md b/README.md index 453fe23..fa16506 100644 --- a/README.md +++ b/README.md @@ -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. diff --git a/rffmpeg.py b/rffmpeg.py index 078f773..ac831d1 100755 --- a/rffmpeg.py +++ b/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() diff --git a/rffmpeg.yml.sample b/rffmpeg.yml.sample index c7fea08..f0f6e79 100644 --- a/rffmpeg.yml.sample +++ b/rffmpeg.yml.sample @@ -27,6 +27,7 @@ rffmpeg: remote: # A YAML list of remote hosts to connect to hosts: + - localhost - gpu1 # - gpu2 # - gpu3 @@ -39,6 +40,9 @@ rffmpeg: args: - "-i" - "/var/lib/jellyfin/id_rsa" + - "-o" + - "StrictHostKeyChecking=no" + # Remote command configuration commands: From e199b5ad68ee80e7bbb286edcb6e68c1056d2e08 Mon Sep 17 00:00:00 2001 From: Bigo Date: Sat, 6 Feb 2021 14:55:36 +0000 Subject: [PATCH 2/2] Remove leftover in config example --- rffmpeg.yml.sample | 2 -- 1 file changed, 2 deletions(-) diff --git a/rffmpeg.yml.sample b/rffmpeg.yml.sample index f0f6e79..b64fb38 100644 --- a/rffmpeg.yml.sample +++ b/rffmpeg.yml.sample @@ -40,8 +40,6 @@ rffmpeg: args: - "-i" - "/var/lib/jellyfin/id_rsa" - - "-o" - - "StrictHostKeyChecking=no" # Remote command configuration