forked from influxdata/influxdb
-
Notifications
You must be signed in to change notification settings - Fork 0
/
stress_client_write.go
112 lines (86 loc) · 2.85 KB
/
stress_client_write.go
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
package stressClient
import (
"bytes"
"fmt"
"io"
"io/ioutil"
"net/http"
"time"
)
// ###############################################
// A selection of methods to manage the write path
// ###############################################
// Packages up Package from channel in goroutine
func (sc *stressClient) spinOffWritePackage(p Package, serv int) {
sc.Add(1)
sc.wc.Increment()
go func() {
sc.retry(p, time.Duration(time.Nanosecond), serv)
sc.Done()
sc.wc.Decrement()
}()
}
// Implements backoff and retry logic for 500 responses
func (sc *stressClient) retry(p Package, backoff time.Duration, serv int) {
// Set Backoff Interval to 500ms
backoffInterval := time.Duration(500 * time.Millisecond)
// Arithmetic backoff for kicks
bo := backoff + backoffInterval
// Make the write request
resp, elapsed, err := sc.prepareWrite(p.Body, serv)
// Find number of times request has been retried
numBackoffs := int(bo/backoffInterval) - 1
// On 500 responses, resp == nil. This logic keeps program for panicing
var statusCode int
if resp == nil {
statusCode = 500
} else {
statusCode = resp.StatusCode
}
// Make a point for reporting
point := sc.writePoint(numBackoffs, p.StatementID, statusCode, elapsed, p.Tracer.Tags, len(p.Body))
// Send the Response(point, tracer)
sc.responseChan <- NewResponse(point, p.Tracer)
// BatchInterval enforcement
bi, _ := time.ParseDuration(sc.wdelay)
time.Sleep(bi)
// Retry if the statusCode was not 204 or the err != nil
if !(statusCode == 204) || err != nil {
// Increment the *Tracer waitgroup if we are going to retry the request
p.Tracer.Add(1)
// Log the error if there is one
fmt.Println(err)
// Backoff enforcement
time.Sleep(bo)
sc.retry(p, bo, serv)
}
}
// Prepares to send the POST request
func (sc *stressClient) prepareWrite(points []byte, serv int) (*http.Response, time.Duration, error) {
// Construct address string
var writeTemplate string
if sc.ssl {
writeTemplate = "https://%v/write?db=%v&precision=%v&u=%v&p=%v"
} else {
writeTemplate = "http://%v/write?db=%v&precision=%v&u=%v&p=%v"
}
address := fmt.Sprintf(writeTemplate, sc.addresses[serv], sc.database, sc.precision, sc.username, sc.password)
// Start timer
t := time.Now()
resp, err := makePost(address, bytes.NewBuffer(points))
elapsed := time.Since(t)
return resp, elapsed, err
}
// Send POST request, read it, and handle errors
func makePost(url string, points io.Reader) (*http.Response, error) {
resp, err := http.Post(url, "text/plain", points)
if err != nil {
return resp, fmt.Errorf("Error making write POST request\n error: %v\n url: %v\n", err, url)
}
body, _ := ioutil.ReadAll(resp.Body)
if resp.StatusCode != 204 {
return resp, fmt.Errorf("Write returned non-204 status code\n StatusCode: %v\n InfluxDB Error: %v\n", resp.StatusCode, string(body))
}
resp.Body.Close()
return resp, nil
}