lib/ferret_server.rb

   1 require 'drb'
   2 require 'thread'
   3 require 'yaml'
   4 require 'erb'
   5
   6 ################################################################################
   7 module ActsAsFerret
   8   module Remote
   9
  10     ################################################################################
  11     class Config
  12
  13       ################################################################################
  14       DEFAULTS = {
  15         'host'      => 'localhost',
  16         'port'      => '9009',
  17         'cf'        => "#{RAILS_ROOT}/config/ferret_server.yml",
  18         'pid_file'  => "#{RAILS_ROOT}/log/ferret_server.pid",
  19         'log_file'  => "#{RAILS_ROOT}/log/ferret_server.log",
  20         'log_level' => 'debug',
  21         'socket'    => nil,
  22       }
  23
  24       ################################################################################
  25       # load the configuration file and apply default settings
  26       def initialize (file=DEFAULTS['cf'])
  27         @everything = YAML.load(ERB.new(IO.read(file)).result)
  28         raise "malformed ferret server config" unless @everything.is_a?(Hash)
  29         @config = DEFAULTS.merge(@everything[RAILS_ENV] || {})
  30         if @everything[RAILS_ENV]
  31           @config['uri'] = socket.nil? ? "druby://#{host}:#{port}" : "drbunix:#{socket}"
  32         end
  33       end
  34
  35       ################################################################################
  36       # treat the keys of the config data as methods
  37       def method_missing (name, *args)
  38         @config.has_key?(name.to_s) ? @config[name.to_s] : super
  39       end
  40
  41     end
  42
  43     #################################################################################
  44     # This class acts as a drb server listening for indexing and
  45     # search requests from models declared to 'acts_as_ferret :remote => true'
  46     #
  47     # Usage:
  48     # - modify RAILS_ROOT/config/ferret_server.yml to suit your needs.
  49     # - environments for which no section in the config file exists will use
  50     #   the index locally (good for unit tests/development mode)
  51     # - run script/ferret_server to start the server:
  52     # script/ferret_server -e production start
  53     # - to stop the server run
  54     # script/ferret_server -e production stop
  55     #
  56     class Server
  57
  58       #################################################################################
  59       # FIXME include detection of OS and include the correct file
  60       require 'unix_daemon'
  61       include(ActsAsFerret::Remote::UnixDaemon)
  62
  63       ################################################################################
  64       cattr_accessor :running
  65
  66       ################################################################################
  67       def initialize
  68         @cfg = ActsAsFerret::Remote::Config.new
  69         ActiveRecord::Base.allow_concurrency = true
  70         ActiveRecord::Base.logger = @logger = Logger.new(@cfg.log_file)
  71         ActiveRecord::Base.logger.level = Logger.const_get(@cfg.log_level.upcase) rescue Logger::DEBUG
  72       end
  73
  74       ################################################################################
  75       # start the server as a daemon process
  76       def start
  77         raise "ferret_server not configured for #{RAILS_ENV}" unless (@cfg.uri rescue nil)
  78         platform_daemon { run_drb_service }
  79       end
  80
  81       ################################################################################
  82       # run the server and block until it exits
  83       def run
  84         raise "ferret_server not configured for #{RAILS_ENV}" unless (@cfg.uri rescue nil)
  85         run_drb_service
  86       end
  87
  88       def run_drb_service
  89         $stdout.puts("starting ferret server...")
  90         self.class.running = true
  91         DRb.start_service(@cfg.uri, self)
  92         DRb.thread.join
  93       rescue Exception => e
  94         @logger.error(e.to_s)
  95         raise
  96       end
  97
  98       #################################################################################
  99       # handles all incoming method calls, and sends them on to the LocalIndex
 100       # instance of the correct model class.
 101       #
 102       # Calls are not queued atm, so this will block until the call returned.
 103       #
 104       def method_missing(name, *args)
 105         @logger.debug "\#method_missing(#{name.inspect}, #{args.inspect})"
 106         retried = false
 107         with_class args.shift do |clazz|
 108           reconnect_when_needed(clazz) do
 109             # using respond_to? here so we not have to catch NoMethodError
 110             # which would silently catch those from deep inside the indexing
 111             # code, too...
 112             if clazz.aaf_index.respond_to?(name)
 113               clazz.aaf_index.send name, *args
 114             elsif clazz.respond_to?(name)
 115               @logger.debug "no luck, trying to call class method instead"
 116               clazz.send name, *args
 117             else
 118               raise NoMethodError.new("method #{name} not supported by DRb server")
 119             end
 120           end
 121         end
 122       rescue => e
 123         @logger.error "ferret server error #{$!}\n#{$!.backtrace.join "\n"}"
 124         raise e
 125       end
 126
 127       # make sure we have a versioned index in place, building one if necessary
 128       def ensure_index_exists(class_name)
 129         @logger.debug "DRb server: ensure_index_exists for class #{class_name}"
 130         with_class class_name do |clazz|
 131           dir = clazz.aaf_configuration[:index_dir]
 132           unless File.directory?(dir) && File.file?(File.join(dir, 'segments')) && dir =~ %r{/\d+(_\d+)?$}
 133             rebuild_index(clazz)
 134           end
 135         end
 136       end
 137
 138       # disconnects the db connection for the class specified by class_name
 139       # used only in unit tests to check the automatic reconnection feature
 140       def db_disconnect!(class_name)
 141         with_class class_name do |clazz|
 142           clazz.connection.disconnect!
 143         end
 144       end
 145
 146       # hides LocalIndex#rebuild_index to implement index versioning
 147       def rebuild_index(clazz, *models)
 148         with_class clazz do |clazz|
 149           models = models.flatten.uniq.map(&:constantize)
 150           models << clazz unless models.include?(clazz)
 151           index = new_index_for(clazz, models)
 152           reconnect_when_needed(clazz) do
 153             @logger.debug "DRb server: rebuild index for class(es) #{models.inspect} in #{index.options[:path]}"
 154             index.index_models models
 155           end
 156           new_version = File.join clazz.aaf_configuration[:index_base_dir], Time.now.utc.strftime('%Y%m%d%H%M%S')
 157           # create a unique directory name (needed for unit tests where
 158           # multiple rebuilds per second may occur)
 159           if File.exists?(new_version)
 160             i = 0
 161             i+=1 while File.exists?("#{new_version}_#{i}")
 162             new_version << "_#{i}"
 163           end
 164
 165           File.rename index.options[:path], new_version
 166           clazz.index_dir = new_version
 167         end
 168       end
 169
 170
 171       protected
 172
 173         def with_class(clazz, *args)
 174           clazz = clazz.constantize if String === clazz
 175           yield clazz, *args
 176         end
 177
 178         def reconnect_when_needed(clazz)
 179           retried = false
 180           begin
 181             yield
 182           rescue ActiveRecord::StatementInvalid => e
 183             if e.message =~ /MySQL server has gone away/
 184               if retried
 185                 raise e
 186               else
 187                 @logger.info "StatementInvalid caught, trying to reconnect..."
 188                 clazz.connection.reconnect!
 189                 retried = true
 190                 retry
 191               end
 192             else
 193               @logger.error "StatementInvalid caught, but unsure what to do with it: #{e}"
 194               raise e
 195             end
 196           end
 197         end
 198
 199         def new_index_for(clazz, models)
 200           aaf_configuration = clazz.aaf_configuration
 201           ferret_cfg = aaf_configuration[:ferret].dup
 202           ferret_cfg.update :auto_flush  => false,
 203                             :create      => true,
 204                             :field_infos => ActsAsFerret::field_infos(models),
 205                             :path        => File.join(aaf_configuration[:index_base_dir], 'rebuild')
 206           returning Ferret::Index::Index.new(ferret_cfg) do |i|
 207             i.batch_size = aaf_configuration[:reindex_batch_size]
 208             i.logger = @logger
 209           end
 210         end
 211
 212     end
 213   end
 214 end