在清洗日志时，有一个应用场景，就是TCP输出时，需要在一个主机挂了的情况下，自已切换到下一个可用入口，而原tcp output仅支持单个目标主机设定。故本人在原tcp的基础上，开发出tcp_multihost输出插件，来满足此场景。
插件在一开始的时候会随机选择一个链路，而在链路出错连续超过３(默认)次后会尝试数组中下一个主机
github: http://github.com/xiaohelong2005
Logstash版本:1.4.2
文件位置:
1# encoding: utf-8 2require "logstash/outputs/base" 3require "logstash/namespace" 4require "thread" 5#@auhor:xiaohelong 6#@date:2014-10-24 7#@email:xiaohelong2005@gmail.com 8# Write events over a TCP socket. 9#Auto select the host from iplist to tolerate the link 10# Each event json is separated by a newline. 11# 12# Can connect to a server, 13class LogStash::Outputs::Tcp_Multihost < LogStash::Outputs::Base 14 config_name "tcp_multihost" 15 milestone 0 16 default :codec, "json" 17 # the address to connect to. 18 #hosts config example 19 config :hosts, :validate => :array, :required => true,:default=>{} 20 config :reconnect_times, :validate => :number,:default=>3 21 # When connect failed,retry interval in sec. 22 config :reconnect_interval, :validate => :number, :default => 10 23 #last available host 24 @hosts_size=0 25 #last available host ip 26@host="0.0.0.0" 27#last available port 28@port=9200 29#retry action count,if retry_count<=0,we need to update the host and port , also include retry_count itself 30@retry_count=0 31#get the desgined index data 32@initloc=0 33 34 public 35 def register 36 require "stud/try" 37 #here we use the recorded host,if recorded host is not available, we need update it 38 @retry_count=@reconnect_times 39 @hosts_size=@hosts.length 40 @logger.info("length:#@hosts_size; hosts:#@hosts") 41 @initloc=Random.rand(@hosts_size)#generate 0-(hosts_size-1) int 42 @logger.info("initloc:#@initloc") 43 icount=0; 44 @hosts.each do |hosthash| 45 @logger.info("hosthash info",hosthash) 46 end#do 47 @host=@hosts[@initloc].keys[0] 48 @port=@hosts[@initloc][@host] 49 50 client_socket = nil 51 @codec.on_event do |payload| 52 begin 53 @retry_count=@reconnect_times#here we need to init retry mark 54 client_socket = connect unless client_socket 55 r,w,e = IO.select([client_socket], [client_socket], [client_socket], nil) 56 # don't expect any reads, but a readable socket might 57 # mean the remote end closed, so read it and throw it away. 58 # we'll get an EOFError if it happens. 59 client_socket.sysread(16384) if r.any? 60 # Now send the payload 61 client_socket.syswrite(payload) if w.any? 62 @logger.info("tcp output info:", :host => @host, :port => @port, 63 :exception => e, :backtrace => e.backtrace) 64 rescue => e 65 @logger.warn("tcp output exception", :host => @host, :port => @port, 66 :exception => e, :backtrace => e.backtrace) 67 client_socket.close rescue nil 68 client_socket = nil 69 @retry_count-=1 70 @logger.info("retry_count:#@retry_count") 71 if @retry_count<=0 72 @initloc+=1 73 @initloc=@initloc%@hosts_size #update init location 74 @host=@hosts[@initloc].keys[0] 75 @port=@hosts[@initloc][@host] 76 @retry_count=@reconnect_times #update retry_count 77 @logger.info("retry_count <=0,initloc:#@initloc,retry_count=#@retry_count:", :host => @host, :port => @port, :exception => e, :backtrace => e.backtrace) 78 end 79 sleep @reconnect_interval 80 retry 81 end 82 end 83 end # def register 84 85 private 86 def connect 87 Stud::try do 88 return TCPSocket.new(@host,@port) 89 end 90 end # def connect 91 public 92 def receive(event) 93 return unless output?(event) 94 @codec.encode(event) 95 end # def receive 96end # class LogStash::Outputs::Tcp_multihost
配置说明(我放在LOGSTASH_HOME/config)：
output{
tcp_multihost{
hosts=>[
{"127.0.0.1"=>"9202"},
{"localhost"=>"9201"},
{"127.0.0.1"=>"9203"},
{"127.0.0.1"=>"9204"}
] #主机列表
workers =>16 #线程,默认1
1reconnect_times=>3 # 默认3次, 尝试多少次数后切换 2 3reconnect_interval=>3 #默认10秒，失败重连间隔 4 5}
}
调用执行:
"LOGSTASH_HOME/bin/logstash" agent --debug -f "LOGSTASH_HOME/config/shipper.config" --pluginpath "LOGSTASH_HOME"
NC接收端可以尝试:
nc -lkv 9201 之类的