Gather and send stats from varnish to stathat

varnish2stathat.go

package main

import (
  "encoding/csv"
	"flag"
	"fmt"
	"io"
	"log"
	"os/exec"
	"sort"
	"strconv"
	"time"

	"github.com/stathat/go"
)

type Stat struct {
	Key   string
	Count int
	Value float64
}

func (s Stat) isCounter() bool {
	return s.Count != 0
}

func watchVarnishLog(stats chan Stat) {
	path, err := exec.LookPath("varnishncsa")
	cmd := exec.Command(path, "-F", "%{Varnish:hitmiss}x,%{Varnish:time_firstbyte}x,%s,%m")

	stdout, err := cmd.StdoutPipe()
	if err != nil {
		log.Fatal(err)
	}
	if err := cmd.Start(); err != nil {
		log.Fatal(err)
	}

	reader := csv.NewReader(stdout)
	for {
		record, err := reader.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Fatal(err)
		}
		if t, err := strconv.ParseFloat(record[1], 64); err == nil {
			stats <- Stat{Key: "testing:" + "time to first byte for " + record[0], Value: t}
		} else {
			log.Printf("could not parse %v as float64", record[1])
		}

		status := record[2]
		stats <- Stat{Key: "testing:" + status, Count: 1}
	}
	if err := cmd.Wait(); err != nil {
		log.Fatal(err)
	}
}

func post(counters map[string]int, values map[string]float64) {
	for key, count := range counters {
		if count > 0 {
			log.Printf("posting ez count: %v %v", key, count)
			if err := stathat.PostEZCount(key, "[email protected]", count); err != nil {
				log.Printf("error posting count %v: %d", err, count)
			}
		}
	}
	for key, value := range values {
		log.Printf("posting ez value: %v %v", key, value)
		if err := stathat.PostEZValue(key, "[email protected]", value); err != nil {
			log.Printf("error posting value %v: %d", err, value)
		}
	}
}

func main() {
	flushInterval := flag.Int64("flush-interval", 10, "Flush interval")
	upper := flag.Int("upper", 90, "percentile for upper stat")
	flag.Parse()

	stats := make(chan Stat, 1000)
	go watchVarnishLog(stats)

	counters := make(map[string]int)
	values := make(map[string][]float64)

	t := time.NewTicker(time.Duration(*flushInterval) * time.Second)
	for {
		select {
		case <-t.C:
			counters_interval := make(map[string]int)
			for key, count := range counters {
				counters_interval[key] = count
				counters[key] = 0
			}
			value_stats := make(map[string]float64)
			for key, value := range values {
				count := len(value)
				if count > 0 {
					sort.Float64s(value)
					sum, max := float64(0), float64(0)
					for i := 0; i < count; i++ {
						sum += value[i]
						if value[i] > max {
							max = value[i]
						}
					}
					mean := sum / float64(count)
					value_stats[key+" (mean)"] = mean
					value_stats[key+" (max)"] = max
					upper_key := fmt.Sprintf("%s (upper %d)", key, *upper)
					value_stats[upper_key] = value[((100-*upper)/100)*count]
					values[key] = value[0:0]
				}
			}
			go post(counters_interval, value_stats)
		case s := <-stats:
			if s.isCounter() {
				counters[s.Key] += s.Count
			} else {
				values[s.Key] = append(values[s.Key], s.Value)
			}
		}
	}
}