stores (seeks) rewritten
This commit is contained in:
Denis Knauf
parent
7a26a9dfad
commit
3c86d43bcf
|
|
@ -14,24 +14,28 @@ module LogAn
|
|||
end
|
||||
|
||||
class SID0 < Command
|
||||
def initialize store, config, sid = 0
|
||||
@store, @config = store, config
|
||||
class <<self
|
||||
attr_accessor :store, :config
|
||||
end
|
||||
|
||||
def initialize sid = 0
|
||||
super sid
|
||||
self[9] = method :event_hostname
|
||||
self[10] = method :event_filerotated
|
||||
end
|
||||
|
||||
def event_filerotated line, sock
|
||||
sid, d = line.unpack 'Na8'
|
||||
@store[ :seeks, sid] = d
|
||||
sid, inode, seek = line.unpack 'NNN'
|
||||
@@store[ :seeks][ sid] = [inode, seek]
|
||||
end
|
||||
|
||||
def event_hostname line, sock
|
||||
@config[ :hosts].each do |sid,host|
|
||||
@@config[ :hosts].each do |sid, host|
|
||||
next unless line == host
|
||||
file = @config[ :files, sid]
|
||||
file = @@config[ :files][ sid]
|
||||
next unless file
|
||||
# command, SID, (inode, seek), file
|
||||
pc = [1, sid, @store[ :seeks, sid] || "\x00\x00\x00\x00\x00\x00\x00\x00", file].pack 'nNa8a*'
|
||||
pc = [1, sid, @@store[ :seeks][ sid], file].pack 'nNNNa*'
|
||||
sock.write [pc.length, pc].pack( 'Na*')
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -12,19 +12,14 @@ module LogAn
|
|||
end
|
||||
|
||||
def seeks read
|
||||
inode, seek = (@@store[ :seeks, @sid] || "\0\0\0\0\0\0\0\0").unpack 'a4N'
|
||||
@@store[ :seeks, @sid] = [inode, read + seek].pack( 'a4N')
|
||||
inode, seek = @@store[ :seeks][@sid]
|
||||
@@store[ :seeks][@sid] = [inode, read + seek]
|
||||
end
|
||||
end
|
||||
|
||||
class Line
|
||||
extend Base
|
||||
attr_reader :sid, :delimiter, :buffer, :linebuffer
|
||||
@@fileparser = []
|
||||
|
||||
def self.[] sid
|
||||
@@fileparser[sid] ||= self.new sid
|
||||
end
|
||||
|
||||
def initialize sid, delimiter = nil
|
||||
@sid, @delimiter = sid, delimiter || "\n"
|
||||
|
|
|
|||
|
|
@ -9,16 +9,25 @@ require 'logan/cache'
|
|||
|
||||
module LogAn::Inc
|
||||
class Main < RobustServer
|
||||
# Open Config.
|
||||
def config env, db, type = nil, flags = nil, &e
|
||||
$stderr.puts "Open Database \"sids.cnf\" #{db.inspect} (#{type.inspect})"
|
||||
def cache db, type, &e
|
||||
type ||= 1+4
|
||||
ret = env[ 'sids.cnf', db, :flags => flags || SBDB::RDONLY]
|
||||
ret = LogAn::AutoValueConvertHash.new ret, &e if type&4 > 0 or e
|
||||
ret = LogAn::Cache.new ret, type&3 if type&3 > 0
|
||||
ret
|
||||
end
|
||||
|
||||
# Open Store.
|
||||
def store env, db, type = nil, flags = nil, &e
|
||||
$stderr.puts [:store, :open, "sids.cnf", db, type].inspect
|
||||
cache env[ 'sids.cnf', db, :flags => flags || SBDB::CREATE | SBDB::AUTO_COMMIT], type, &e
|
||||
end
|
||||
|
||||
# Open Config.
|
||||
def config env, db, type = nil, flags = nil, &e
|
||||
$stderr.puts [:config, :open, "sids.cnf", db, type].inspect
|
||||
cache env[ 'sids.cnf', db, :flags => flags || SBDB::RDONLY], type, &e
|
||||
end
|
||||
|
||||
# Prepare Server.
|
||||
#
|
||||
# * conf:
|
||||
|
|
@ -40,6 +49,7 @@ module LogAn::Inc
|
|||
|
||||
# Open Loglines-databases
|
||||
@logs = LogAn::Loglines.new @conf[:logs]
|
||||
LogAn::Inc::FileParser::Base.logdb = @logs
|
||||
|
||||
# Open config-databases
|
||||
Dir.mkdir @conf[:etc] rescue Errno::EEXIST
|
||||
|
|
@ -47,18 +57,23 @@ module LogAn::Inc
|
|||
log_config: SBDB::Env::LOG_IN_MEMORY | SBDB::Env::LOG_AUTO_REMOVE,
|
||||
flags: SBDB::CREATE | SBDB::Env::INIT_TXN | Bdb::DB_INIT_MPOOL)
|
||||
|
||||
# Set inc-config - stored in etc/inc.cnf
|
||||
# Open configs
|
||||
begin
|
||||
inc = @conf[:inc] = {}
|
||||
%w[hosts files fileparser].each {|key| @conf[:inc][key.to_sym] = config( @etc, key) }
|
||||
inc[:fileparser] = config( @etc, 'fileparser') {|val| Safebox.run "[#{val}]"}
|
||||
configs = @conf[:configs] = {}
|
||||
%w[hosts files].each {|key| configs[key.to_sym] = config( @etc, key) }
|
||||
configs[:fileparser] = config( @etc, 'fileparser') {|val| Safebox.run "[#{val}]"}
|
||||
end
|
||||
|
||||
# Open seeks-database
|
||||
begin
|
||||
stores = @conf[:stores] = {}
|
||||
stores[:seeks] = store( @etc, 'seeks', 3,
|
||||
lambda {|val| val.pack( 'NN') }) {|val| (val||0.chr*8).unpack( 'NN') }
|
||||
LogAn::Inc::FileParser::Base.store = LogAn::Inc::SID0.store = stores
|
||||
end
|
||||
@store = LogAn::Cache.new LogAn::AutoValueConvertHash.new( @etc[ 'sids.store', 'seeks', SBDB::Recno, SBDB::CREATE | SBDB::AUTO_COMMIT]), 3
|
||||
|
||||
# Prepare Inc-server - create server
|
||||
LogAn::Inc::FileParser::Base.logdb = @logs
|
||||
LogAn::Inc::FileParser::Base.store = @store
|
||||
@serv = LogAn::Inc::Server.new :sock => TCPServer.new( *@conf[:server]), :config => @conf[:inc]
|
||||
@serv = LogAn::Inc::Server.new :sock => TCPServer.new( *@conf[:server]), :config => @conf[:configs]
|
||||
|
||||
# Shutdown on signals
|
||||
@sigs[:INT] = @sigs[:TERM] = method( :shutdown)
|
||||
|
|
|
|||
|
|
@ -98,7 +98,7 @@ class LogAn::Inc::Server < ::Select::Server
|
|||
def event_cmd cmd
|
||||
sid, line = cmd.unpack 'Na*'
|
||||
fps = @config[:fileparser][sid]
|
||||
fp.event_line line, self if fileparser
|
||||
fp.event_line line, self if fp
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
Loading…
Reference in a new issue