<commit_before>require 'bosh/dev/sandbox/database_migrator'

module Bosh::Dev::Sandbox
  class DirectorService

    REPO_ROOT = File.expand_path('../../../../../../', File.dirname(__FILE__))
    ASSETS_DIR = File.expand_path('bosh-dev/assets/sandbox', REPO_ROOT)
    DIRECTOR_UUID = 'deadbeef'

    DEFAULT_DIRECTOR_CONFIG = 'director_test.yml'
    DIRECTOR_CONF_TEMPLATE = File.join(ASSETS_DIR, 'director_test.yml.erb')

    DIRECTOR_PATH = File.expand_path('bosh-director', REPO_ROOT)

    def initialize(director_port, redis_port, base_log_path, director_tmp_path, director_config, logger)
      @redis_port = redis_port
      @logger = logger
      @director_tmp_path = director_tmp_path
      @director_config = director_config

      log_location = "#{base_log_path}.director.out"
      @process = Service.new(
        %W[bosh-director -c #{@director_config}],
        {output: log_location},
        @logger,
      )

      @socket_connector = SocketConnector.new('director', 'localhost', director_port, log_location, @logger)

      @worker_processes = 3.times.map do |index|
        Service.new(
          %W[bosh-director-worker -c #{@director_config}],
          {output: "#{base_log_path}.worker_#{index}.out", env: {'QUEUE' => '*'}},
          @logger,
        )
      end

      @database_migrator = DatabaseMigrator.new(DIRECTOR_PATH, @director_config, @logger)
    end

    def start(config)
      write_config(config)

      migrate_database

      reset

      @process.start

      start_workers

      begin
        # CI does not have enough time to start bosh-director
        # for some parallel tests; increasing to 60 secs (= 300 tries).
        @socket_connector.try_to_connect(300)
      rescue
        output_service_log(@process)
        raise
      end
    end

    def stop
      stop_workers
      @process.stop
    end

    private

    def migrate_database
      unless @database_migrated
        @database_migrator.migrate
      end

      @database_migrated = true
    end

    def start_workers
      @worker_processes.each(&:start)
      start_time = Time.now
      timeout = 60 * 5
      until resque_is_ready?
        if (Time.now - start_time) > timeout
           raise "Resque failed to start workers in #{timeout} seconds"
        end

        sleep 0.5
      end
    end

    def stop_workers
      @logger.debug('Waiting for Resque queue to drain...')
      start_time = Time.now
      timeout = 60
      until resque_is_done?
        if (Time.now - start_time) > timeout
          raise "Resque queue failed to drain in #{timeout} seconds"
        end

        sleep 0.1
      end
      @logger.debug('Resque queue drained')

      Redis.new(host: 'localhost', port: @redis_port).flushdb

      # wait for resque workers in parallel for fastness
      @worker_processes.map { |worker_process| Thread.new { worker_process.stop } }.each(&:join)
    end

    def reset
      FileUtils.rm_rf(@director_tmp_path)
      FileUtils.mkdir_p(@director_tmp_path)
      File.open(File.join(@director_tmp_path, 'state.json'), 'w') do |f|
        f.write(Yajl::Encoder.encode('uuid' => DIRECTOR_UUID))
      end
    end<commit_msg>Rely on number of attempts instead of timeout when waiting for workers<commit_after>require 'bosh/dev/sandbox/database_migrator'

module Bosh::Dev::Sandbox
  class DirectorService

    REPO_ROOT = File.expand_path('../../../../../../', File.dirname(__FILE__))
    ASSETS_DIR = File.expand_path('bosh-dev/assets/sandbox', REPO_ROOT)
    DIRECTOR_UUID = 'deadbeef'

    DEFAULT_DIRECTOR_CONFIG = 'director_test.yml'
    DIRECTOR_CONF_TEMPLATE = File.join(ASSETS_DIR, 'director_test.yml.erb')

    DIRECTOR_PATH = File.expand_path('bosh-director', REPO_ROOT)

    def initialize(director_port, redis_port, base_log_path, director_tmp_path, director_config, logger)
      @redis_port = redis_port
      @logger = logger
      @director_tmp_path = director_tmp_path
      @director_config = director_config

      log_location = "#{base_log_path}.director.out"
      @process = Service.new(
        %W[bosh-director -c #{@director_config}],
        {output: log_location},
        @logger,
      )

      @socket_connector = SocketConnector.new('director', 'localhost', director_port, log_location, @logger)

      @worker_processes = 3.times.map do |index|
        Service.new(
          %W[bosh-director-worker -c #{@director_config}],
          {output: "#{base_log_path}.worker_#{index}.out", env: {'QUEUE' => '*'}},
          @logger,
        )
      end

      @database_migrator = DatabaseMigrator.new(DIRECTOR_PATH, @director_config, @logger)
    end

    def start(config)
      write_config(config)

      migrate_database

      reset

      @process.start

      start_workers

      begin
        # CI does not have enough time to start bosh-director
        # for some parallel tests; increasing to 60 secs (= 300 tries).
        @socket_connector.try_to_connect(300)
      rescue
        output_service_log(@process)
        raise
      end
    end

    def stop
      stop_workers
      @process.stop
    end

    private

    def migrate_database
      unless @database_migrated
        @database_migrator.migrate
      end

      @database_migrated = true
    end

    def start_workers
      @worker_processes.each(&:start)
      attempt = 0
      delay = 0.5
      timeout = 60 * 5
      max_attempts = timeout/delay

      until resque_is_ready?
        if attempt > max_attempts
           raise "Resque failed to start workers in #{timeout} seconds"
        end

        attempt += 1
        sleep delay
      end
    end

    def stop_workers
      @logger.debug('Waiting for Resque queue to drain...')
      attempt = 0
      delay = 0.1
      timeout = 60
      max_attempts = timeout/delay

      until resque_is_done?
        if attempt > max_attempts
          raise "Resque queue failed to drain in #{timeout} seconds"
        end

        attempt += 1
        sleep delay
      end
      @logger.debug('Resque queue drained')

      Redis.new(host: 'localhost', port: @redis_port).flushdb

      # wait for resque workers in parallel for fastness
      @worker_processes.map { |worker_process| Thread.new { worker_process.stop } }.each(&:join)
    end

    def reset
      FileUtils.rm_rf(@director_tmp_path)
      FileUtils.mkdir_p(@director_tmp_path)
      File.open(File.join(@director_tmp_path, 'state.json'), 'w') do |f|
        f.write(Yajl::Encoder.encode('uuid' => DIRECTOR_UUID))
      end
    end
{'fraction_non_alphanumeric': 0.10161662817551963, 'fraction_numerical': 0.006158583525789068, 'mean_word_length': 2.526601520086862, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, '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': '371187', 'n_tokens_mistral': 2131, 'n_tokens_neox': 1996, 'n_words': 446}