#!/usr/bin/env ruby

# Starts N instances of EZQ::Processor. N is pulled from key
# 'number_of_processes' in userdata.yml. Valid values are integer N:N>1 and 'auto'.
# A value of 'auto' will start as many processes as there are available cores
# (according to the OS)

require 'yaml'
require 'parallel'
require 'logger'
require 'fileutils'
require 'net/http'
require 'sys/filesystem'
require 'ezq/utils/userdata'
require 'aws-sdk'

def substitute(command,idx)
  dbp = @debug_port + idx
  exp = @exit_port + idx
  command += " --log ./processor_" + "%02d" % idx + ".log"
  command += " --log_severity info"
  command += " --queue #{@rec_queue}" if !@rec_queue.empty?
  command += " --error_queue #{@err_queue}" if !@err_queue.empty?
  command += " --error_topic #{@err_topic}" if !@err_topic.empty?
  command += " --result_queue #{@res_queue}" if !@res_queue.empty?
  command += " --token #{@loggly_token}" if @loggly_token
  command += " --loggly_severity #{@loggly_level}" if @loggly_level
  command += " --app_name #{@app_name}" if @app_name
  command += " --debug_port #{dbp}" if @pry_server
  command += " --exit_port #{exp}"

  return command
end


def start


  stagger = ARGV.shift.to_i  # nil.to_i always returns 0

  log = Logger.new('processor_fan_out.log')

  run = true
  Signal.trap('SIGTERM'){run = false }

  command = 'processor -c receive_queue_config.yml'

  userdata = ''
  begin
    userdata = EZQ::UserData.load
  rescue => e
    warn "Error opening or parsing 'userdata.yml': #{e}. Aborting"
    exit(1)
  end

  num = userdata.fetch('number_of_processes',1)
  num = num == 'auto' ?
        Parallel.processor_count :
        num.to_i

  @rec_queue = userdata.fetch('receive_queue_name','')
  puts "Overriding receive queue setting with #{@rec_queue}" if !@rec_queue.empty?
  @err_queue = userdata.fetch('error_queue_name','')
  @err_topic = userdata.fetch('error_topic','')
  @res_queue = userdata.fetch('result_queue_name','')
  @loggly_token = userdata.fetch('loggly_token',nil)
  @loggly_level = userdata.fetch('loggly_level',nil)
  @app_name = userdata.fetch('app_name',nil)
  @pry_server = userdata.fetch('pry_server',false)
  @debug_port = 7755
  @exit_port = 8642

  sandbox = userdata.fetch('sandbox', false)
  all_files = Dir.entries('.') - ['.', '..']

  pids = {}

  num.times do |idx|
    cmd = substitute(command,idx)
    log.info "Initial start of process #{idx}"

    dirname = '.'
    if sandbox
      log.info "  --Creating sandbox directory for number #{idx}"
      dirname = "sandbox_#{idx}"
      begin
        Dir.mkdir(dirname)
      rescue
        # blank on purpose
      end

      begin
        log.info "  --Copying files into sandbox directory"
        FileUtils.cp_r(all_files, dirname)
      rescue
        # blank on purpose
      end
    end

    log.info "  --Starting #{cmd}"
    pid = spawn(cmd,:chdir=>dirname)
    pids[pid] = idx
    sleep stagger
  end

  # Check in on the processes every 30s and restart as needed
  loop do
    sleep(30)

    # If we've received SIGTERM, ask all subprocesses to end, wait for them to
    # do so, and then exit
    if !run
      # Send TERM to all children
      pids.keys do |pid|
        Process.kill("TERM", pid)
      end
      # Wait for children to finish
      while pids.size > 0
        sleep(5)
        puts "Checking #{pids.size} child processes for termination"
        pids.keys.each do |pid|
          puts "Checking pid #{pid}"
          begin
            if Process.wait(pid, Process::WNOHANG)
              puts "#{pid} has exited"
              pids.delete(pid)
            else
              puts "#{pid} has not exited yet"
            end
          rescue Errno::ECHILD
              puts "Deleting pid #{pid} because we got Errno::ECHILD"
              pids.delete(pid)
          end
        end
      end
      exit(0)
    end

    pids.to_a.each do |pid,idx|
      exited = Process.wait(pid,Process::WNOHANG)
      if exited
        pids.delete(pid)
        cmd = substitute(command,idx)
        exit_status = $?.exitstatus
        log.info "Detected failure with status '#{$?.to_i}'. Restarting '#{cmd}'"
        dirname = sandbox ? "sandbox_#{idx}" : '.'
        ensure_free_space(dirname)
        if run && !termination_requested && (exit_status != 0) # Don't try to restart if proc exited due to term warning OR if proc exited with success status
          pid = spawn(cmd, :chdir=>dirname)
          pids[pid] = idx
          sleep stagger
        end
      end
    end

    # 2021-08-25 Commenting this out since it won't play well with running all of this as
    # a system service on Linux. It's what we want to do in the general case, but not when
    # running as a service, since the service's stop request already indicates we're being
    # controlled by a uid that has the power to terminate the instance.
    # # If all processes have exited with success, we should attempt to terminate the instance
    # if pids.empty?
    #   uri = URI.parse("http://169.254.169.254/latest/meta-data/instance-id")
    #   @instance_id = Net::HTTP.get_response(uri).body
    #   @instance = Aws::EC2::Instance.new(@instance_id)
    #   @instance.terminate
    # end
  end

end


# WARNING: AWS-specific
def termination_requested
  # This URL req will succeed ONLY if termination has been requested
  uri = URI.parse("http://169.254.169.254/latest/meta-data/spot/instance-action")
  Net::HTTP.get_response(uri).value # Raises if resp code is not 2XX
  return true
rescue
  return false
end


def ensure_free_space(dirname)
  s = Sys::Filesystem.stat(Dir.pwd)
  gb_free = s.block_size * s.blocks_available / (1024.0 ** 3)
  if gb_free < 1  # Have we dropped below 1 GB?
    # This is a bit hacky. We're going to delete sets of files and directories
    # that we know ahead of time might exist in this dir. Ideally we would
    # stop *all* processors, back up out of the run dir, delete everything, pull
    # down fresh archives and basically start over. That strategy has the following
    # downsides:
    # 1. It contains a LOT of steps, so there are lots of places for stuff to go wrong
    # 2. In the case of machines running lots of processors, it means signalling to
    #    all of them to gracefully exit. Doable, but not trivial (esp. on Windows).
    # 3. Due to the way Windows handles dangling file handles, you can't safely
    #    delete the file you're currently running. That would mean pushing the process
    #    monitor up a directory, which complicates environment and dir handling.
    #
    # So, we go the slightly hacky way for now: delete files that match
    # specific glob patterns.
    globs = ["md", "enterprise", "*.in", "*.message", "out*.txt", "*.pdf"]
    globs.each do |glob|
      fs = Dir.glob(File.join(dirname, glob))
      fs.each{ |f| FileUtils.rm_rf(f) }
    end
  end
end


begin
  start()
rescue Interrupt
  puts "Killed by user. Aborting."
end
