-
Notifications
You must be signed in to change notification settings - Fork 1.3k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(influxdb): queue batches to the influxdb if it's slowing down
Previously to this k6 will write to influxdb every second, but if that write took more than 1 second it won't start a second write but instead the wait for it. This will generally lead to the write times getting bigger and bigger as more and more data is being written until the max body that influxdb will take is reached when it will return an error and k6 will drop that data. With this commit there will be a configurable number of parallel writes (10 by default) that will trigger again every 1 second (also now configurable), but if those get exhausted it will start queueing the samples each second instead of combining them and than writing a big chunk which has a chance of hitting the max body. I tested with a simple script doing batch request for an empty local file with 40VUs. Without an output it was getting 8.1K RPS with 650MB of memory usage. previous to this commit the usage of ram was ~5.7GB for 5736 rps and practically all the data gets lost if you don't up the max body and even than a lot of the data is lost while the memory usage goes up. After this commit the usage of ram was ~2.4GB(or less in some of the tests) with 6273 RPS and there was no lost of data. Even with this commit doing 2 hour of that simple script dies after 1hour and 35 minutes using around 15GB (the test system has 16). Can't be sure of lost of data, as influxdb eat 32GB of memory trying to visualize it. Some minor problems with this solution is that: 1. We use a lot of goroutines if things start slowing down - probably not a big problem 2. We can probably better batch stuff if we add/keep all the unsend samples together 3. By far the biggest: because the writes are slow if the test is stopped (with Ctrl+C) or it finishes naturally, waiting for those writes can take considerable amount of time - in the above example the 4 minutes tests generally took around 5 minutes :( All of those can be better handled with some more sophisticated queueing code at later time. closes #1081, fixes #1100, fixes #182
- Loading branch information
Showing
4 changed files
with
179 additions
and
21 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,118 @@ | ||
| package influxdb | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "io" | ||
| "net" | ||
| "net/http" | ||
| "sync" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/loadimpact/k6/stats" | ||
| "github.com/stretchr/testify/require" | ||
| null "gopkg.in/guregu/null.v3" | ||
| ) | ||
|
|
||
| func TestBadConcurrentWrites(t *testing.T) { | ||
| c := NewConfig() | ||
| t.Run("0", func(t *testing.T) { | ||
| c.ConcurrentWrites = null.IntFrom(0) | ||
| _, err := New(*c) | ||
| require.Error(t, err) | ||
| require.Equal(t, err.Error(), "influxdb's ConcurrentWrites must be a possitive number") | ||
| }) | ||
|
|
||
| t.Run("-2", func(t *testing.T) { | ||
| c.ConcurrentWrites = null.IntFrom(-2) | ||
| _, err := New(*c) | ||
| require.Error(t, err) | ||
| require.Equal(t, err.Error(), "influxdb's ConcurrentWrites must be a possitive number") | ||
| }) | ||
|
|
||
| t.Run("2", func(t *testing.T) { | ||
| c.ConcurrentWrites = null.IntFrom(2) | ||
| _, err := New(*c) | ||
| require.NoError(t, err) | ||
| }) | ||
| } | ||
|
|
||
| func testCollectorCycle(t testing.TB, handler http.HandlerFunc, body func(testing.TB, *Collector)) { | ||
| s := &http.Server{ | ||
| Addr: ":", | ||
| Handler: handler, | ||
| MaxHeaderBytes: 1 << 20, | ||
| } | ||
| l, err := net.Listen("tcp", "127.0.0.1:0") | ||
| require.NoError(t, err) | ||
| defer func() { | ||
| _ = l.Close() | ||
| }() | ||
|
|
||
| defer func() { | ||
| require.NoError(t, s.Shutdown(context.Background())) | ||
| }() | ||
|
|
||
| go func() { | ||
| require.Equal(t, http.ErrServerClosed, s.Serve(l)) | ||
| }() | ||
|
|
||
| config := NewConfig() | ||
| config.Addr = null.StringFrom("http://" + l.Addr().String()) | ||
| c, err := New(*config) | ||
| require.NoError(t, err) | ||
|
|
||
| require.NoError(t, c.Init()) | ||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| var wg sync.WaitGroup | ||
| defer cancel() | ||
| wg.Add(1) | ||
| go func() { | ||
| defer wg.Done() | ||
| c.Run(ctx) | ||
| }() | ||
|
|
||
| body(t, c) | ||
|
|
||
| cancel() | ||
| wg.Wait() | ||
| } | ||
| func TestCollector(t *testing.T) { | ||
| var samplesRead int | ||
| defer func() { | ||
| require.Equal(t, samplesRead, 20) | ||
| }() | ||
| testCollectorCycle(t, func(rw http.ResponseWriter, r *http.Request) { | ||
| var b = bytes.NewBuffer(nil) | ||
| _, _ = io.Copy(b, r.Body) | ||
| for { | ||
| s, err := b.ReadString('\n') | ||
| if len(s) > 0 { | ||
| samplesRead++ | ||
| } | ||
| if err != nil { | ||
| break | ||
| } | ||
| } | ||
|
|
||
| rw.WriteHeader(204) | ||
| }, func(tb testing.TB, c *Collector) { | ||
| var samples = make(stats.Samples, 10) | ||
| for i := 0; i < len(samples); i++ { | ||
| samples[i] = stats.Sample{ | ||
| Metric: stats.New("testGauge", stats.Gauge), | ||
| Time: time.Now(), | ||
| Tags: stats.NewSampleTags(map[string]string{ | ||
| "something": "else", | ||
| "VU": "21", | ||
| "else": "something", | ||
| }), | ||
| Value: 2.0, | ||
| } | ||
| } | ||
| c.Collect([]stats.SampleContainer{samples}) | ||
| c.Collect([]stats.SampleContainer{samples}) | ||
| }) | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters