<commit_before>module Synapse
  class Haproxy
    attr_reader :opts
    def initialize(opts)
      super()

      %w{global defaults reload_command}.each do |req|
        raise ArgumentError, "haproxy requires a #{req} section" if !opts.has_key?(req)
      end

      req_pairs = {
        'do_writes' => 'config_file_path',
        'do_socket' => 'socket_file_path',
        'do_reloads' => 'reload_command'}

      req_pairs.each do |cond, req|
        if opts[cond]
          raise ArgumentError, "the `#{req}` option is required when `#{cond}` is true" unless opts[req]
        end
      end

      @opts = opts
    end

    def update_config(watchers)
      new_config = generate_config(watchers)

      updated = write_config(new_config) if @opts['do_writes']
      restart if (updated && @opts['do_reloads'])
    end

    # generates a new config based on the state of the watchers
    def generate_config(watchers)
      new_config = generate_base_config + "\n"
      new_config << watchers.map {|w| generate_listen_stanza(w)}.join("\n")

      log.debug "synapse: new haproxy config: #{new_config}"
      return new_config
    end

    # generates the global and defaults sections of the config file
    def generate_base_config
      base_config = "# auto-generated by synapse at #{Time.now}\n"

      %w{global defaults}.each do |section|
        base_config << "\n#{section}\n"
        @opts[section].each do |option|
          base_config << "\t#{option}\n"
        end
      end

      if @opts['extra_sections']
        @opts['extra_sections'].each do |title, section|
          base_config << "\n#{title}\n"
          section.each do |option|
            base_config << "\t#{option}\n"
          end
        end
      end

      return base_config
    end

    # generates an individual stanza for a particular watcher
    def generate_listen_stanza(watcher)
      if watcher.backends.empty?
        log.warn "synapse: no backends found for watcher #{watcher.name}"
        return ""
      end

      stanza = "listen #{watcher.name} localhost:#{watcher.local_port}\n"

      watcher.listen.each do |line|
        stanza << "\t#{line}\n"
      end

      watcher.backends.shuffle.each do |backend|
        stanza << "\tserver #{backend['name']} #{backend['host']}:#{backend['port']} #{watcher.server_options}\n" 
      end

      return stanza
    end

    # writes the config
    def write_config(new_config)
      begin 
        old_config = File.read(@opts['config_file_path'])
      rescue Errno::ENOENT => e
        log.info "synapse: could not open haproxy config file at #{@opts['config_file_path']}"
        old_config = ""
      end

      if old_config == new_config
        return false
      else
        File.open(@opts['config_file_path'],'w') {|f| f.write(new_config)}
        return true
      end
    end

    # restarts haproxy
    def restart
      res = `#{opts['reload_command']}`.chomp
      raise "failed to reload haproxy via #{opts['reload_command']}: #{res}" unless $?.success?
    end
  end
end<commit_msg>use haproxy stats socket to enable/disable backends when possible<commit_after>require 'socket'

module Synapse
  class Haproxy
    attr_reader :opts
    def initialize(opts)
      super()

      %w{global defaults reload_command}.each do |req|
        raise ArgumentError, "haproxy requires a #{req} section" if !opts.has_key?(req)
      end

      req_pairs = {
        'do_writes' => 'config_file_path',
        'do_socket' => 'socket_file_path',
        'do_reloads' => 'reload_command'}

      req_pairs.each do |cond, req|
        if opts[cond]
          raise ArgumentError, "the `#{req}` option is required when `#{cond}` is true" unless opts[req]
        end
      end

      @opts = opts
      @restart_required = true
    end

    def update_config(watchers)
      # if we don't already require a restart, try just updating the backends
      unless @restart_required
        update_backends(watchers) if @opts['do_socket']
      end

      # generate a new config
      new_config = generate_config(watchers)

      # if we write config files, lets do that and then possibly restart
      if @opts['do_writes']
        write_config(new_config)
        restart if @opts['do_reloads'] && @restart_required
      end
    end

    # generates a new config based on the state of the watchers
    def generate_config(watchers)
      new_config = generate_base_config + "\n"
      new_config << watchers.map {|w| generate_listen_stanza(w)}.join("\n")

      log.debug "synapse: new haproxy config: #{new_config}"
      return new_config
    end

    # generates the global and defaults sections of the config file
    def generate_base_config
      base_config = "# auto-generated by synapse at #{Time.now}\n"

      %w{global defaults}.each do |section|
        base_config << "\n#{section}\n"
        @opts[section].each do |option|
          base_config << "\t#{option}\n"
        end
      end

      if @opts['extra_sections']
        @opts['extra_sections'].each do |title, section|
          base_config << "\n#{title}\n"
          section.each do |option|
            base_config << "\t#{option}\n"
          end
        end
      end

      return base_config
    end

    # generates an individual stanza for a particular watcher
    def generate_listen_stanza(watcher)
      if watcher.backends.empty?
        log.warn "synapse: no backends found for watcher #{watcher.name}"
        return ""
      end

      stanza = "listen #{watcher.name} localhost:#{watcher.local_port}\n"

      watcher.listen.each do |line|
        stanza << "\t#{line}\n"
      end

      watcher.backends.shuffle.each do |backend|
        stanza << "\tserver #{backend['name']} #{backend['host']}:#{backend['port']} #{watcher.server_options}\n"
      end

      return stanza
    end

    # tries to set active backends via haproxy's stats socket
    # because we can't add backends via the socket, we might still need to restart haproxy
    def update_backends(watchers)
      # first, get a list of existing servers for various backends
      begin
        s = UNIXSocket.new(@opts['socket_file_path'])
        s.write('show stat;')
        info = s.read()
      rescue StandardError => e
        log.warn "synapse: unhandled error reading stats socket: #{e.inspect}"
        @restart_required = true
        return
      end

      # parse the stats output to get current backends
      cur_backends = {}
      info.split("\n").each do |line|
        next if line[0] == '#'

        parts = line.split(',')
        next if ['FRONTEND', 'BACKEND'].include?(parts[1])

        cur_backends[parts[0]] ||= []
        cur_backends[parts[0]] << parts[1]
      end

      # build a list of backends that should be enabled
      enabled_backends = {}
      watchers.each do |watcher|
        enabled_backends[watcher.name] = []
        next if watcher.backends.empty?

        unless cur_backends.include? watcher.name
          log.debug "synapse: restart required because we added new section #{watcher.name}"
          @restart_required = true
          return
        end

        watcher.backends.each do |backend|
          unless cur_backends[watcher.name].include? backend['name']
            log.debug "synapse: restart required because we have a new backend #{watcher.name}/#{backend['name']}"
            @restart_required = true
            return
          end

          enabled_backends[watcher.name] << backend['name']
        end
      end

      # actually enable the enabled backends, and disable the disabled ones
      cur_backends.each do |section, backends|
        backends.each do |backend|
          if enabled_backends[section].include? backend
            command = "enable server #{section}/#{backend};"
          else
            command = "disable server #{section}/#{backend};"
          end

          # actually write the command to the socket
          begin
            s = UNIXSocket.new(@opts['socket_file_path'])
            s.write(command)
            output = s.read()
          rescue StandardError => e
            log.warn "synapse: unknown error writing to socket"
            @restart_required = true
            return
          else
            unless output == "\n"
              log.warn "synapse: socket command #{command} failed: #{output}"
              @restart_required = true
              return
            end
          end
        end
      end
    end

    # writes the config
    def write_config(new_config)
      begin
        old_config = File.read(@opts['config_file_path'])
      rescue Errno::ENOENT => e
        log.info "synapse: could not open haproxy config file at #{@opts['config_file_path']}"
        old_config = ""
      end

      if old_config == new_config
        return false
      else
        File.open(@opts['config_file_path'],'w') {|f| f.write(new_config)}
        return true
      end
    end

    # restarts haproxy
    def restart
      res = `#{opts['reload_command']}`.chomp
      raise "failed to reload haproxy via #{opts['reload_command']}: #{res}" unless $?.success?
      @restart_required = false
    end
  end
end
{'fraction_non_alphanumeric': 0.1073106764288056, 'fraction_numerical': 0.0005463883728554257, 'mean_word_length': 2.3087490961677513, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5965915', 'n_tokens_mistral': 2723, 'n_tokens_neox': 2672, 'n_words': 756}