/usr/share/gocode/src/github.com/influxdata/influxdb/stress/v2/stress_client/stress_client.go is in golang-github-influxdb-influxdb-dev 1.1.1+dfsg1-4.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 | package stressClient
import (
"strings"
"sync"
)
// Type refers to the different Package types
type Type int
// There are two package types, Write and Query
const (
Write Type = iota
Query
)
func startStressClient(packageCh <-chan Package, directiveCh <-chan Directive, responseCh chan<- Response, testID string) {
c := &stressClient{
testID: testID,
addresses: []string{"localhost:8086"},
ssl: false,
username: "",
password: "",
precision: "ns",
database: "stress",
startDate: "2016-01-01",
qdelay: "0s",
wdelay: "0s",
wconc: 10,
qconc: 5,
packageChan: packageCh,
directiveChan: directiveCh,
responseChan: responseCh,
}
// start listening for writes and queries
go c.listen()
// start listening for state changes
go c.directiveListen()
}
type stressClient struct {
testID string
// State for the Stress Test
addresses []string
precision string
startDate string
database string
wdelay string
qdelay string
username string
password string
ssl bool
// Channels from statements
packageChan <-chan Package
directiveChan <-chan Directive
// Response channel
responseChan chan<- Response
// Concurrency utilities
sync.WaitGroup
sync.Mutex
// Concurrency Limit for Writes and Reads
wconc int
qconc int
// Manage Read and Write concurrency seperately
wc *ConcurrencyLimiter
rc *ConcurrencyLimiter
}
// NewTestStressClient returns a blank stressClient for testing
func newTestStressClient(url string) (*stressClient, chan Directive, chan Package) {
pkgChan := make(chan Package)
dirChan := make(chan Directive)
pe := &stressClient{
testID: "foo_id",
addresses: []string{url},
precision: "s",
startDate: "2016-01-01",
database: "fooDatabase",
wdelay: "50ms",
qdelay: "50ms",
ssl: false,
username: "",
password: "",
wconc: 5,
qconc: 5,
packageChan: pkgChan,
directiveChan: dirChan,
wc: NewConcurrencyLimiter(1),
rc: NewConcurrencyLimiter(1),
}
return pe, dirChan, pkgChan
}
// stressClient starts listening for Packages on the main channel
func (sc *stressClient) listen() {
defer sc.Wait()
sc.wc = NewConcurrencyLimiter(sc.wconc)
sc.rc = NewConcurrencyLimiter(sc.qconc)
l := NewConcurrencyLimiter((sc.wconc + sc.qconc) * 2)
counter := 0
for p := range sc.packageChan {
l.Increment()
go func(p Package) {
defer l.Decrement()
switch p.T {
case Write:
sc.spinOffWritePackage(p, (counter % len(sc.addresses)))
case Query:
sc.spinOffQueryPackage(p, (counter % len(sc.addresses)))
}
}(p)
counter++
}
}
// Set handles all SET requests for test state
func (sc *stressClient) directiveListen() {
for d := range sc.directiveChan {
sc.Lock()
switch d.Property {
// addresses is a []string of target InfluxDB instance(s) for the test
// comes in as a "|" seperated array of addresses
case "addresses":
addr := strings.Split(d.Value, "|")
sc.addresses = addr
// percison is the write precision for InfluxDB
case "precision":
sc.precision = d.Value
// writeinterval is an optional delay between batches
case "writeinterval":
sc.wdelay = d.Value
// queryinterval is an optional delay between the batches
case "queryinterval":
sc.qdelay = d.Value
// database is the InfluxDB database to target for both writes and queries
case "database":
sc.database = d.Value
// username for the target database
case "username":
sc.username = d.Value
// username for the target database
case "password":
sc.password = d.Value
// use https if sent true
case "ssl":
if d.Value == "true" {
sc.ssl = true
}
// concurrency is the number concurrent writers to the database
case "writeconcurrency":
conc := parseInt(d.Value)
sc.wconc = conc
sc.wc.NewMax(conc)
// concurrentqueries is the number of concurrent queriers database
case "queryconcurrency":
conc := parseInt(d.Value)
sc.qconc = conc
sc.rc.NewMax(conc)
}
d.Tracer.Done()
sc.Unlock()
}
}
|