Merge pull request #95 from joekiller/queue_stats_output

Queue stats output

gor_stat.go

@@ -0,0 +1,67 @@
+package main
+
+import (
+	"time"
+	"log"
+	"strconv"
+)
+
+const (
+	rate = 5
+)
+
+type GorStat struct {
+	statName	string
+	latest		int
+	mean	 	int
+	max			int
+	count		int
+}
+
+func NewGorStat(statName string) (s *GorStat) {
+	s = new(GorStat)
+	s.statName = statName
+	s.latest = 0
+	s.mean = 0
+	s.max = 0
+	s.count = 0
+
+	if Settings.stats {
+		log.Println(s.statName + ":latest,mean,max,count,count/second")
+		go s.reportStats()
+	}
+	return
+}
+
+func (s *GorStat) Write(latest int) {
+	if Settings.stats {
+		if latest > s.max {
+			s.max = latest
+		}
+		if latest != 0 {
+			s.mean = (s.mean + latest) / 2
+		}
+		s.latest = latest
+		s.count = s.count + 1
+	}
+}
+
+func (s *GorStat) Reset() {
+	s.latest = 0
+	s.max = 0
+	s.mean = 0
+	s.count = 0
+}
+
+func (s *GorStat) String() string {
+	return s.statName + ":" + strconv.Itoa(s.latest) + "," + strconv.Itoa(s.mean) + "," + strconv.Itoa(s.max) + "," + strconv.Itoa(s.count) + "," + strconv.Itoa(s.count / rate)
+}
+
+func (s *GorStat) reportStats() {
+	for {
+			log.Println(s)
+			s.Reset()
+			time.Sleep(rate * time.Second)
+	}
+}
+

output_http.go

@@ -46,6 +46,8 @@ type HTTPOutput struct {
 	methods HTTPMethods
 
 	elasticSearch *es.ESPlugin
+
+	bufStats *GorStat
 }
 
 func NewHTTPOutput(options string, headers HTTPHeaders, methods HTTPMethods, elasticSearchAddr string) io.Writer {
@@ -63,6 +65,7 @@ func NewHTTPOutput(options string, headers HTTPHeaders, methods HTTPMethods, ela
 	o.methods = methods
 
 	o.buf = make(chan []byte, 100)
+	o.bufStats = NewGorStat("output_http")
 
 	if elasticSearchAddr != "" {
 		o.elasticSearch = new(es.ESPlugin)
@@ -100,6 +103,7 @@ func (o *HTTPOutput) Write(data []byte) (n int, err error) {
 	copy(buf, data)
 
 	o.buf <- buf
+	o.bufStats.Write(len(o.buf))
 
 	return len(data), nil
 }

output_tcp.go

@@ -13,6 +13,7 @@ type TCPOutput struct {
 	address string
 	limit   int
 	buf     chan []byte
+	bufStats *GorStat
 }
 
 func NewTCPOutput(options string) io.Writer {
@@ -22,6 +23,7 @@ func NewTCPOutput(options string) io.Writer {
 	o.address = optionsArr[0]
 
 	o.buf = make(chan []byte, 100)
+	o.bufStats = NewGorStat("output_tcp")
 
 	if len(optionsArr) > 1 {
 		o.limit, _ = strconv.Atoi(optionsArr[1])
@@ -52,6 +54,7 @@ func (o *TCPOutput) Write(data []byte) (n int, err error) {
 	data = append(data,[]byte("¶")...)
 	copy(new_buf, data)
 	o.buf <- new_buf
+	o.bufStats.Write(len(o.buf))
 
 	return len(data), nil
 }

settings.go

@@ -13,6 +13,7 @@ const (
 
 type AppSettings struct {
 	verbose bool
+	stats bool
 
 	splitOutput bool
 
@@ -45,6 +46,7 @@ func init() {
 	flag.Usage = usage
 
 	flag.BoolVar(&Settings.verbose, "verbose", false, "Turn on verbose/debug output")
+	flag.BoolVar(&Settings.stats, "stats", false, "Turn on queue stats output")
 
 	flag.BoolVar(&Settings.splitOutput, "split-output", false, "By default each output gets same traffic. If set to `true` it splits traffic equally among all outputs.")