Ruby module to assist in capturing stdout/stderr from a detached thread

A ruby module to assist in capturing stdout and stderr from a detached thread in jRuby.md

A ruby module to assist in capturing stdout/stderr from a detached thread in jRuby

1. create some temp files (but not Tempfiles since they would be GC'd too quickly)
2. append standard file descriptor redirects to the commands that we'll run in the detached spawned process, i.e. 1>>stdout_tempfile_path 2>>stderr_tempfile_path
3. tack on a final command that will trigger copying the files to S3 regardless of exit status of original commands, i.e. {original commands with io redirection &&'d together}; bundle exec rake cleanup_task
4. cleanup other io redirection temp files older than some threshold

Example usage:

job_identifier = "install_and_run_cowsay"
uuid = SecureRandom.hex
commands = [
   "cd #{Dir.pwd}",
   IoRedirect.command_with_io_redirection("apt-get install cowsay", job_identifier, uuid),
   IoRedirect.command_with_io_redirection("cowsay I Love nix", job_identifier, uuid)
].join(" && ") + "; bundle exec rake io_redirect:copy_logs_to_s3[#{job_identifier},#{uuid}]"

pid = Process.spawn(commands) # => 12345
thread = Process.detach(pid) # => #<Thread:0x60e1d87c run>

io_redirect.rake

require_relative "io_redirect"

namespace :io_redirect do
  desc "Copy IoRedirect logs to s3"
  task :copy_logs_to_s3, [:job_identifier, :uuid] do |_task, args|
    begin
       puts "Uploading IoRedirect logs for #{args.job_identifier} and #{args.uuid}"
       IoRedirect.copy_logs_to_s3(args.job_identifier, args.uuid)
    ensure
      threshold = 1.day.ago.utc
      puts "Purging IoRedirect logs older than #{threshold}"
      IoRedirect.purge_logs(threshold)
    end
  end
end

io_redirect.rb

module IoRedirect
  extend self

  DIRECTORY_TAG = "io_redirect"
  STDOUT_FILE = "stdout.log"
  STDERR_FILE = "stderr.log"

  def command_with_io_redirection(command, job_identifier, uuid)
    stdout_path = create_log(job_identifier, uuid, STDOUT_FILE)
    stderr_path = create_log(job_identifier, uuid, STDERR_FILE)
    "#{command} 1>>#{stdout_path} 2>>#{stderr_path}"
  end

  def create_log(job_identifier, uuid, file_descriptor)
    base_path = log_base_path(job_identifier, uuid)
    log_path = File.join(base_path, file_descriptor)
    FileUtils.mkdir_p(base_path) unless Dir.exist?(base_path)
    FileUtils.touch(log_path) && File.truncate(log_path, 0)
    log_path
  end

  def log_base_path(job_identifier, uuid)
    File.join(Config.log_path, [job_identifier, DIRECTORY_TAG, uuid].join('_'))
  end

  def find_logs(job_identifier, uuid)
    [STDOUT_FILE, STDERR_FILE].inject({}) do |log_paths, file_descriptor|
      log_path = File.join(log_base_path(job_identifier, uuid), file_descriptor)
      log_paths[file_descriptor] = log_path if File.exist?(log_path)
      log_paths
    end
  end

  def purge_logs(threshold_time)
    Dir.glob(File.join(Config.log_path, "*_#{DIRECTORY_TAG}_*")).each do |log_path_match|
      if File.directory?(log_path_match) && File.stat(log_path_match).mtime < threshold_time
        FileUtils.rmtree(log_path_match)
      end
    end
  end

  def copy_logs_to_s3(job_identifier, uuid)
    logs = find_logs(job_identifier, uuid)
    logs.each do |log_file_name, log_file_path|
      key = File.join(Config.s3_key_base, job_identifier, uuid, log_file_name)
      Aws::S3::Object.new(
        bucket_name: Config.s3_bucket,
        key: key,
        client: Config.s3_client
      ).upload_file(log_file_path)
    end
  end
end