kumofs 0.4.13-6 / usr / bin / kumolog

#!/usr/bin/env ruby
#
# kumofs
#
# Copyright (C) 2009 FURUHASHI Sadayuki
#
#    Licensed under the Apache License, Version 2.0 (the "License");
#    you may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.
#

class StringSource
	def initialize(src)
		@src = src
	end

	def read(off, size)
		if @src.length < off + size
			throw :finish
		end
		@src[off, size]
	end

	def free_until(off)
	end
end

class StreamSource
	def initialize(io)
		@io = io
		@from = 0
		@buffer = ''
		@rbuf = ''
	end

	def read(off, size)
		while @from + @buffer.length < off + size
			@buffer << @io.sysread(1024, @rbuf)
		end
		@buffer[off - @from, size]
	rescue Exception
		throw :finish
	end

	def free_until(off)
		free = off - @from
		if free > 0
			@buffer.slice!(0, free)
			@from += free
		end
	end
end


def fixstr(code)
	r = ""
	8.times {|i|
		c = ((code >> (8*(7-i))) & 0xff)
		r << c.chr if c != 0
	}
	r
end

def type_check(obj)
	obj.is_a?(Array) && obj.length == 3 &&
		obj[0].is_a?(Numeric) && obj[1].is_a?(Numeric) &&
		obj[2].is_a?(Hash)
end


def do_recover(src, off)
	puts "recover at #{off}"
	pk = MessagePack::Unpacker.new

	while true
		br = src.read(off,4).unpack('N')[0]
		doff = off + 4

		begin
			pk.reset
			rl = pk.execute(src.read(doff,br), 0)

			if pk.finished && rl == br && type_check(pk.data)
				return off
			end
		rescue
			off += 1
		end
	end
end


def do_parse(src, skip, count, &block)
	count = (1<<31) if count < 0
	catch(:finish) {
		off = 0
		rl = 0
	
		pk = MessagePack::Unpacker.new
	
		while count > 0
			br = src.read(off,4).unpack('N')[0]
			off += 4
	
			begin
				pk.reset
				rl = pk.execute(src.read(off,br), 0)
	
				unless pk.finished? && rl == br && type_check(pk.data)
					raise "format error"
				end
	
				obj = pk.data
	
				hash = {}
				name = fixstr(obj[0])
				version = obj[1]
				obj[2].each_pair {|k,v|
					hash[fixstr(k)] = v
				}
	
			rescue
				off = do_recover(src, off-3)
				next
			end
	
			off += br
			src.free_until(off)
	
			if skip > 0
				skip -= 1
			else
				block.call(name, version, hash)
				count -= 1
			end
		end
	}
end


require 'optparse'

arg = {
	:follow => false,
	:tail   => false,
	:head   => false,
	:line   => 10,
}

op = OptionParser.new
op.on('-t', '--tail')   {|b| arg[:tail]  = b }
op.on('-h', '--head')   {|b| arg[:head]  = b }
op.on('-f', '--follow') {|b| arg[:follow] = b }
op.on('-n [-]N', '--lines', Integer) {|n| arg[:line] = n }
op.banner += " <logfile.mpac>"

op.parse!(ARGV)

if ARGV.length != 1
	puts op.to_s
	exit 1
end


fname = ARGV[0]
tail = nil
head = nil
skip  = 0
count = -1

if arg[:tail]
	if (n = arg[:line]) > 0
		tail = Array.new(n)
	else
		skip = -n
	end
elsif arg[:head]
	if (n = arg[:line]) > 0
		count = n
	else
		head = Array.new(-n)
	end
end


if arg[:follow]
	begin
		require 'shellwords'
		fname = Shellwords.escape(fname)
	rescue
		fname = "'#{fname}'"
	end
	stream = IO.popen("tail -f #{fname}")
elsif fname == "-"
	stream = $stdin
else
	stream = File.open(fname)
end

src = StreamSource.new(stream)
#src = StringSource.new(File.read(fname))



begin
	require 'rubygems'
rescue LoadError
end
require 'yaml'
require 'pp'
require 'msgpack'

class Hash
	def hmap(&block)
		m = {}
		each_pair {|k, v|
			m[k] = block.call(k, v)
		}
		m
	end
end

conf = YAML.load DATA.read.gsub(/(^\t+)/) {
	'  ' * $+.length
}

@msgdb = conf["message"]

@filterdb = conf["filter"].hmap {|name, hash|
	hash.hmap {|key, proc|
		[ proc[0], eval("lambda{|val|#{proc[1]}}") ]
	}
}


def print_log(name, version, hash)
	msg = @msgdb[name] || "#{name}.#{version}"

	if filter = @filterdb["#{name}.#{version}"]
		filter.each_pair {|key, proc|
			val = hash.delete(key)
			begin
				hash[proc[0]] = proc[1].call(val)
			rescue
			end
		}
	end

	vals = hash.map {|k, v|
		pv = v.pretty_inspect.rstrip
		pv = v if pv[1..-2] == v
		"#{k}=[#{pv}]"
	}.sort_by{|kv| kv[0] }.join('  ')

	puts "%s.%s  %-15s  %s" % [name, version, msg, vals]
end


if tail
	do_parse(src, skip, count) {|*params|
		tail.shift
		tail.push params
	}
	tail.each(&method(:print_log))

elsif head
	do_parse(src, skip, count) {|*params|
		head.shift
		head.push params
		if h = head[0]
			print_log(h)
		end
	}

else
	do_parse(src, skip, count, &method(:print_log))
end


__END__
proc:
	- &addr |
		require 'socket'
		if val.length == 6
			addr = Socket.pack_sockaddr_in(0, '0.0.0.0')
			addr[2,6] = val[0,6]
		else
			addr = Socket.pack_sockaddr_in(0, '::')
			addr[2,2]  = val[0,2]
			addr[8,20] = val[2,20]
		end
		Socket.unpack_sockaddr_in(addr).reverse.join(':')

	- &time |
		Time.at(val).strftime("%Y-%m-%d %H:%M:%S")

	- &clocktime |
		Time.at(val>>32).strftime("%Y-%m-%d %H:%M:%S") + " clock #{val & 0xffffffff}"

message:
	SM: Manager start
	SS: Server start
	SW: Gateway start
	eP: unknown partner
	nS: new server
	lS: lost server
	ers: replicate-set failed
	erd: replicate-delete failed
	eg: Get failed
	es: Set failed
	ed: Delete failed

filter:
	SM.2:
		time: [time, *time]
		addr: [address, *addr]
		Padd: [partner, *addr]
	SS.2:
		time: [time, *time]
		addr: [address, *addr]
		db:   [database, val]
		mgr1: [manager1, *addr]
		mgr2: [manager2, *addr]
		sadd: [stream_listen, *addr]
		tmpd: [tmp_dir, val]
		bkup: [backup_prefix, val]
	SW.2:
		time: [time, *time]
		mgr1: [manager1, *addr]
		mgr2: [manager2, *addr]
	eP.2:
		addr: [address, *addr]
	nS.2:
		addr: [address, *addr]
	lS.2:
		addr: [address, *addr]
	ers.3:
		to:   [to, *addr]
		cktm: [clocktime, *clocktime]
	erd.3:
		to:   [to, *addr]
	eP.3:
		addr: [address, *addr]
		time: [time, *time]
	nS.3:
		addr: [address, *addr]
		time: [time, *time]
	lS.3:
		addr: [address, *addr]
		time: [time, *time]
	eg.3:
		time: [time, *time]
	es.3:
		time: [time, *time]
	ed.3:
		time: [time, *time]
	ers.4:
		to:   [to, *addr]
		cktm: [clocktime, *clocktime]
		time: [time, *time]
	erd.4:
		to:   [to, *addr]
		time: [time, *time]