capella 0.1.3

An asynchronous StatsD clone
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

package main

import (
	"bytes"
	"flag"
	"fmt"
	"io/ioutil"
	"log"
	"math/rand"
	"net"
	"os"
	"sync"
	"sync/atomic"
	"time"
)

const metricFileName = "sample_metrics.txt"

var (
	// Command-line flags.
	batchSize   int
	concurrency int
	capellaAddr string
	duration    time.Duration

	// A set of metrics to chose from.
	metrics map[int]string

	metricCount uint64
	wg          sync.WaitGroup
)

func init() {
	rand.Seed(time.Now().Unix())

	flag.IntVar(&batchSize, "b", 100, "the number of metrics to batch before sending")
	flag.IntVar(&concurrency, "c", 10, "the number of concurrent connections")
	flag.StringVar(&capellaAddr, "a", "127.0.0.1:8125", "the address of the capella instance with the port")
	flag.DurationVar(&duration, "d", 30*time.Second, "how long the benchmark will last")
}

func readMetricsFile() error {
	b, err := ioutil.ReadFile(metricFileName)
	if err != nil {
		return err
	}

	metrics = make(map[int]string)
	sep := []byte("\n")
	index := 0

	// We use an integer index to pick a metric at random later.
	for _, m := range bytes.Split(b, sep) {
		metric := string(m)
		if metric == "" {
			continue
		}
		metrics[index] = metric
		index += 1
	}

	return nil
}

// Get a random metric from the map that was parsed from file earlier.
func getRandomMetric() string {
	i := rand.Intn(len(metrics))
	return metrics[i]
}

// A worker function kicked off by main.
func work() {
	defer wg.Done()

	// Setup the UDP connection before starting a timer.
	conn, err := net.Dial("udp", capellaAddr)
	if err != nil {
		fmt.Fprintf(os.Stderr, "error connecting to %s: %s\n", capellaAddr, err)
		return
	}

	buffer := bytes.NewBufferString("")
	batchCounter := 0

	timer := time.After(duration)
	for {
		select {
		case <-timer:
			conn.Close()
			return
		default:
			atomic.AddUint64(&metricCount, 1)
			batchCounter += 1
			buffer.WriteString(getRandomMetric() + "\n")
			if batchCounter == batchSize {
				_, err = fmt.Fprint(conn, buffer.String())
				if err != nil {
					fmt.Fprintf(os.Stderr, "error writing to socket: %s\n", err)
				}
				batchCounter = 0
				buffer.Reset()
			}
		}
	}
}

func printSummary() {
	mc := atomic.LoadUint64(&metricCount)
	fmt.Printf("metrics sent: %d\n", mc)
	fmt.Printf("metrics per second: %.2f\n", float64(mc)/duration.Seconds())
}

func main() {
	flag.Parse()

	// Read in the sample metrics file.
	err := readMetricsFile()
	if err != nil {
		log.Fatalf("error reading in metrics file: %s\n", err)
	}

	wg.Add(concurrency)
	for i := 0; i < concurrency; i++ {
		go work()
	}
	wg.Wait()

	printSummary()
}