s2l -> s2l | rotate | bdb
This commit is contained in:
parent
896f641856
commit
1efdf76ffc
90
bin/s2l.rb
90
bin/s2l.rb
|
@ -7,6 +7,8 @@ require 'uuidtools'
|
||||||
require 'socket'
|
require 'socket'
|
||||||
require 'select'
|
require 'select'
|
||||||
require 'robustserver'
|
require 'robustserver'
|
||||||
|
require 'active_support'
|
||||||
|
require 'syslog2logan/rotate'
|
||||||
|
|
||||||
$logger = Logger.new $stderr
|
$logger = Logger.new $stderr
|
||||||
|
|
||||||
|
@ -19,17 +21,13 @@ class S2L < Select::Server
|
||||||
end
|
end
|
||||||
|
|
||||||
def event_new_client a
|
def event_new_client a
|
||||||
debug :connection => {:new => a}
|
logger.debug :connection => {:new => a}
|
||||||
{ :clientclass => S2L::Socket, :dbs => @dbs }
|
{ :clientclass => S2L::Socket, :dbs => @dbs }
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
module Kernel
|
module Kernel
|
||||||
def debug( *p) $logger.debug *p end
|
def logger() $logger end
|
||||||
def info( *p) $logger.info *p end
|
|
||||||
def warn( *p) $logger.warn *p end
|
|
||||||
def error( *p) $logger.error *p end
|
|
||||||
def fatal( *p) $logger.fatal *p end
|
|
||||||
end
|
end
|
||||||
|
|
||||||
class S2L::Socket < Select::Socket
|
class S2L::Socket < Select::Socket
|
||||||
|
@ -39,105 +37,43 @@ class S2L::Socket < Select::Socket
|
||||||
end
|
end
|
||||||
|
|
||||||
def event_line v
|
def event_line v
|
||||||
debug :line => v
|
logger.debug :line => v
|
||||||
@dbs.emit v
|
@dbs.emit v
|
||||||
end
|
end
|
||||||
alias emit event_line
|
alias emit event_line
|
||||||
end
|
end
|
||||||
|
|
||||||
class Rotate
|
|
||||||
def initialize db, &e
|
|
||||||
@rdb, @env, @dbs = db, db.home, {}
|
|
||||||
self.hash = e || lambda {|k|
|
|
||||||
[k.timestamp.to_i/60/60/24].pack 'N'
|
|
||||||
}
|
|
||||||
end
|
|
||||||
|
|
||||||
def hash= e
|
|
||||||
self.hash &e
|
|
||||||
end
|
|
||||||
|
|
||||||
def hash &e
|
|
||||||
@hash_func = e if e
|
|
||||||
@hash_func
|
|
||||||
end
|
|
||||||
|
|
||||||
def hashing k
|
|
||||||
@hash_func.call k
|
|
||||||
end
|
|
||||||
|
|
||||||
def db_name id
|
|
||||||
h = hashing id
|
|
||||||
n = @rdb[ h]
|
|
||||||
if n
|
|
||||||
n = UUIDTools::UUID.parse_raw n
|
|
||||||
else
|
|
||||||
n = UUIDTools::UUID.timestamp_create
|
|
||||||
@rdb[ h] = n.raw
|
|
||||||
info :create => n.to_s
|
|
||||||
end
|
|
||||||
n
|
|
||||||
end
|
|
||||||
|
|
||||||
def db n
|
|
||||||
@env[ n.to_s, :type => SBDB::Btree, :flags => SBDB::CREATE | SBDB::AUTO_COMMIT]
|
|
||||||
end
|
|
||||||
|
|
||||||
def queue n
|
|
||||||
@env[ "newids.queue", :type => SBDB::Queue, :flags => SBDB::CREATE | SBDB::AUTO_COMMIT, :re_len => 16]
|
|
||||||
end
|
|
||||||
|
|
||||||
def sync
|
|
||||||
@dbs.each{|n,db|db.sync}
|
|
||||||
@rdb.sync
|
|
||||||
end
|
|
||||||
|
|
||||||
def close
|
|
||||||
@dbs.each{|n,db|db.close 0}
|
|
||||||
@rdb.close 0
|
|
||||||
end
|
|
||||||
|
|
||||||
def put v
|
|
||||||
id = UUIDTools::UUID.timestamp_create
|
|
||||||
s = [0x10, v].pack 'Na*'
|
|
||||||
n = db_name id
|
|
||||||
db( n)[ id.raw] = s
|
|
||||||
queue( n).push id.raw
|
|
||||||
end
|
|
||||||
alias emit put
|
|
||||||
end
|
|
||||||
|
|
||||||
class Main < RobustServer
|
class Main < RobustServer
|
||||||
def initialize conf
|
def initialize conf
|
||||||
super
|
super
|
||||||
@logger = $logger
|
@logger = $logger
|
||||||
@conf = conf
|
@conf = conf
|
||||||
info :open => S2L
|
logger.info :open => S2L
|
||||||
@serv = S2L.new :sock => TCPServer.new( *@conf[:server])
|
@serv = S2L.new :sock => TCPServer.new( *@conf[:server])
|
||||||
info :create => {:home => @conf[:home]}
|
logger.info :create => {:home => @conf[:home]}
|
||||||
Dir.mkdir @conf[:home] rescue Errno::EEXIST
|
Dir.mkdir @conf[:home] rescue Errno::EEXIST
|
||||||
@sigs[:INT] = @sigs[:TERM] = method(:shutdown)
|
@sigs[:INT] = @sigs[:TERM] = method(:shutdown)
|
||||||
@sigs[:USR1] = method(:state)
|
@sigs[:USR1] = method(:state)
|
||||||
end
|
end
|
||||||
|
|
||||||
def state s = nil
|
def state s = nil
|
||||||
debug :server => @serv
|
logger.debug :server => @serv
|
||||||
end
|
end
|
||||||
|
|
||||||
def shutdown s = nil
|
def shutdown s = nil
|
||||||
info :shutdown => [s, Signal[s]]
|
logger.info :shutdown => [s, Signal[s]]
|
||||||
@serv.close
|
@serv.close
|
||||||
exit 0
|
exit 0
|
||||||
end
|
end
|
||||||
|
|
||||||
def run
|
def run
|
||||||
info :open => SBDB::Env
|
logger.info :open => SBDB::Env
|
||||||
SBDB::Env.new( @conf[:home],
|
SBDB::Env.new( @conf[:home],
|
||||||
log_config: SBDB::Env::LOG_IN_MEMORY | SBDB::Env::LOG_AUTO_REMOVE,
|
log_config: SBDB::Env::LOG_IN_MEMORY | SBDB::Env::LOG_AUTO_REMOVE,
|
||||||
flags: SBDB::CREATE | SBDB::Env::INIT_TXN | Bdb::DB_INIT_MPOOL) do |dbenv|
|
flags: SBDB::CREATE | SBDB::Env::INIT_TXN | Bdb::DB_INIT_MPOOL) do |dbenv|
|
||||||
info :open => Rotate
|
logger.info :open => Rotate
|
||||||
@serv.dbs = Rotate.new dbenv[ 'rotates.db', :type => SBDB::Btree, :flags => SBDB::CREATE | Bdb::DB_AUTO_COMMIT]
|
@serv.dbs = Rotate.new dbenv[ 'rotates.db', :type => SBDB::Btree, :flags => SBDB::CREATE | Bdb::DB_AUTO_COMMIT]
|
||||||
info :run => @serv
|
logger.info :run => @serv
|
||||||
@serv.run
|
@serv.run
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
@ -145,4 +81,4 @@ end
|
||||||
|
|
||||||
Main.main :home => 'logs', :server => [ '', 1514], :retries => [1,1] # [10, 10]
|
Main.main :home => 'logs', :server => [ '', 1514], :retries => [1,1] # [10, 10]
|
||||||
|
|
||||||
info :halted
|
logger.info :halted
|
||||||
|
|
2
lib/syslog2logan/rotate.rb
Normal file
2
lib/syslog2logan/rotate.rb
Normal file
|
@ -0,0 +1,2 @@
|
||||||
|
|
||||||
|
require 'syslog2logan/rotate/base.rb'
|
58
lib/syslog2logan/rotate/base.rb
Executable file
58
lib/syslog2logan/rotate/base.rb
Executable file
|
@ -0,0 +1,58 @@
|
||||||
|
#!/usr/bin/ruby
|
||||||
|
|
||||||
|
require 'logger'
|
||||||
|
require 'uuidtools'
|
||||||
|
require 'active_support'
|
||||||
|
|
||||||
|
class Rotate
|
||||||
|
# open_db_func: must returns a db-object with #[] and #[]=.
|
||||||
|
# #sync and #close are optional, for Rotate#sync, Rotate#close.
|
||||||
|
def initialize hash_func = nil, &open_db_func
|
||||||
|
@dbs = Hash.new {|h,k| h[k] = open_db_func.call(k) }
|
||||||
|
hash_func ||= lambda {|k| [k.timestamp.to_i/1.day].pack 'N' }
|
||||||
|
define_method :hashing, &hash_func
|
||||||
|
@rotate = @dbs['rotate']
|
||||||
|
@queue = @dbs['queue']
|
||||||
|
end
|
||||||
|
|
||||||
|
def db_name id
|
||||||
|
h = hashing id
|
||||||
|
n = @rdb[ h]
|
||||||
|
if n
|
||||||
|
n = UUIDTools::UUID.parse_raw n
|
||||||
|
else
|
||||||
|
n = UUIDTools::UUID.timestamp_create
|
||||||
|
@rdb[ h] = n.raw
|
||||||
|
logger.info :create => n.to_s
|
||||||
|
end
|
||||||
|
n
|
||||||
|
end
|
||||||
|
|
||||||
|
# Synchronize data to disc.
|
||||||
|
# Only avaible if db-backend provides #sync.
|
||||||
|
def sync
|
||||||
|
@dbs.each {|n, db| db.sync }
|
||||||
|
@rotate.sync
|
||||||
|
@queue.sync
|
||||||
|
end
|
||||||
|
|
||||||
|
# Close databases.
|
||||||
|
# Only avaible if db-backend provides #close.
|
||||||
|
def close
|
||||||
|
@dbs.each {|n, db| db.close }
|
||||||
|
@rotate.close
|
||||||
|
@queue.close
|
||||||
|
end
|
||||||
|
|
||||||
|
# Put new logline to databases.
|
||||||
|
# This will be written in a database with an UUID as name.
|
||||||
|
# If this db don't exist, it will be created via open_db_func (#initialize).
|
||||||
|
def put v
|
||||||
|
id = UUIDTools::UUID.timestamp_create
|
||||||
|
s = [0x10, v].pack 'Na*'
|
||||||
|
n = db_name id
|
||||||
|
@dbs[n][ id.raw] = s
|
||||||
|
@queue.push id.raw
|
||||||
|
end
|
||||||
|
alias emit put
|
||||||
|
end
|
108
lib/syslog2logan/rotate/bdb.rb
Executable file
108
lib/syslog2logan/rotate/bdb.rb
Executable file
|
@ -0,0 +1,108 @@
|
||||||
|
#!/usr/bin/ruby
|
||||||
|
|
||||||
|
require 'logger'
|
||||||
|
require 'rubygems'
|
||||||
|
require 'sbdb'
|
||||||
|
require 'uuidtools'
|
||||||
|
require 'socket'
|
||||||
|
require 'select'
|
||||||
|
require 'robustserver'
|
||||||
|
require 'active_support'
|
||||||
|
|
||||||
|
class Rotate::BDB
|
||||||
|
def initialize db, &e
|
||||||
|
@rdb, @env, @dbs = db, db.home, {}
|
||||||
|
self.hash = e || lambda {|k|
|
||||||
|
[k.timestamp.to_i/1.hour].pack 'N'
|
||||||
|
}
|
||||||
|
end
|
||||||
|
|
||||||
|
def hash= e
|
||||||
|
self.hash &e
|
||||||
|
end
|
||||||
|
|
||||||
|
def hash &e
|
||||||
|
@hash_func = e if e
|
||||||
|
@hash_func
|
||||||
|
end
|
||||||
|
|
||||||
|
def hashing k
|
||||||
|
@hash_func.call k
|
||||||
|
end
|
||||||
|
|
||||||
|
def db_name id
|
||||||
|
h = hashing id
|
||||||
|
n = @rdb[ h]
|
||||||
|
if n
|
||||||
|
n = UUIDTools::UUID.parse_raw n
|
||||||
|
else
|
||||||
|
n = UUIDTools::UUID.timestamp_create
|
||||||
|
@rdb[ h] = n.raw
|
||||||
|
logger.info :create => n.to_s
|
||||||
|
end
|
||||||
|
n
|
||||||
|
end
|
||||||
|
|
||||||
|
def db n
|
||||||
|
@env[ n.to_s, :type => SBDB::Btree, :flags => SBDB::CREATE | SBDB::AUTO_COMMIT]
|
||||||
|
end
|
||||||
|
|
||||||
|
def queue n
|
||||||
|
@env[ "newids.queue", :type => SBDB::Queue, :flags => SBDB::CREATE | SBDB::AUTO_COMMIT, :re_len => 16]
|
||||||
|
end
|
||||||
|
|
||||||
|
def sync
|
||||||
|
@dbs.each {|n, db| db.sync }
|
||||||
|
@rdb.sync
|
||||||
|
end
|
||||||
|
|
||||||
|
def close
|
||||||
|
@dbs.each {|n, db| db.close 0 }
|
||||||
|
@rdb.close 0
|
||||||
|
end
|
||||||
|
|
||||||
|
def put v
|
||||||
|
id = UUIDTools::UUID.timestamp_create
|
||||||
|
s = [0x10, v].pack 'Na*'
|
||||||
|
n = db_name id
|
||||||
|
db( n)[ id.raw] = s
|
||||||
|
queue( n).push id.raw
|
||||||
|
end
|
||||||
|
alias emit put
|
||||||
|
end
|
||||||
|
|
||||||
|
class Main < RobustServer
|
||||||
|
def initialize conf
|
||||||
|
super
|
||||||
|
@logger = $logger
|
||||||
|
@conf = conf
|
||||||
|
logger.info :open => S2L
|
||||||
|
@serv = S2L.new :sock => TCPServer.new( *@conf[:server])
|
||||||
|
logger.info :create => {:home => @conf[:home]}
|
||||||
|
Dir.mkdir @conf[:home] rescue Errno::EEXIST
|
||||||
|
@sigs[:INT] = @sigs[:TERM] = method(:shutdown)
|
||||||
|
@sigs[:USR1] = method(:state)
|
||||||
|
end
|
||||||
|
|
||||||
|
def state s = nil
|
||||||
|
logger.debug :server => @serv
|
||||||
|
end
|
||||||
|
|
||||||
|
def shutdown s = nil
|
||||||
|
logger.info :shutdown => [s, Signal[s]]
|
||||||
|
@serv.close
|
||||||
|
exit 0
|
||||||
|
end
|
||||||
|
|
||||||
|
def run
|
||||||
|
logger.info :open => SBDB::Env
|
||||||
|
SBDB::Env.new( @conf[:home],
|
||||||
|
log_config: SBDB::Env::LOG_IN_MEMORY | SBDB::Env::LOG_AUTO_REMOVE,
|
||||||
|
flags: SBDB::CREATE | SBDB::Env::INIT_TXN | Bdb::DB_INIT_MPOOL) do |dbenv|
|
||||||
|
logger.info :open => Rotate
|
||||||
|
@serv.dbs = Rotate.new dbenv[ 'rotates.db', :type => SBDB::Btree, :flags => SBDB::CREATE | Bdb::DB_AUTO_COMMIT]
|
||||||
|
logger.info :run => @serv
|
||||||
|
@serv.run
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
Loading…
Reference in a new issue