diff --git a/Gopkg.lock b/Gopkg.lock
index 8c60d724f65ca920918d793d07653f6c6b042310..ee398f6f12a4104ccf9f3741af2a49ef895c119b 100644
--- a/Gopkg.lock
+++ b/Gopkg.lock
@@ -904,7 +904,7 @@
[[projects]]
branch = "master"
- digest = "1:5fc7e79142258f1ce47429c00fedb9ee04d8562e7fd477f2d9ae28a692e80969"
+ digest = "1:43edb14ccbe0d18f38bd55511250c39ccc3917c64a514a0e791408ac19a71558"
name = "github.com/prometheus/prometheus"
packages = [
"discovery",
@@ -925,6 +925,7 @@
"pkg/gate",
"pkg/labels",
"pkg/modtimevfs",
+ "pkg/pool",
"pkg/relabel",
"pkg/textparse",
"pkg/timestamp",
@@ -1608,7 +1609,9 @@
"github.com/pkg/errors",
"github.com/prometheus/client_golang/prometheus",
"github.com/prometheus/client_golang/prometheus/promauto",
+ "github.com/prometheus/client_golang/prometheus/promhttp",
"github.com/prometheus/client_golang/prometheus/testutil",
+ "github.com/prometheus/client_model/go",
"github.com/prometheus/common/config",
"github.com/prometheus/common/expfmt",
"github.com/prometheus/common/model",
@@ -1618,6 +1621,7 @@
"github.com/prometheus/prometheus/discovery/targetgroup",
"github.com/prometheus/prometheus/pkg/labels",
"github.com/prometheus/prometheus/pkg/modtimevfs",
+ "github.com/prometheus/prometheus/pkg/pool",
"github.com/prometheus/prometheus/pkg/relabel",
"github.com/prometheus/prometheus/pkg/textparse",
"github.com/prometheus/prometheus/promql",
@@ -1637,8 +1641,9 @@
"github.com/weaveworks/common/user",
"golang.org/x/net/context",
"google.golang.org/grpc",
+ "google.golang.org/grpc/codes",
"google.golang.org/grpc/health/grpc_health_v1",
- "google.golang.org/grpc/metadata",
+ "google.golang.org/grpc/status",
"gopkg.in/alecthomas/kingpin.v2",
"gopkg.in/fsnotify.v1",
"gopkg.in/yaml.v2",
diff --git a/pkg/chunkenc/gzip.go b/pkg/chunkenc/gzip.go
index ff5bc5d54ce22d7e717b63c8be987046fe2226cc..3b3bb67ee35442357201c185e033ca6fc7ba5fdc 100644
--- a/pkg/chunkenc/gzip.go
+++ b/pkg/chunkenc/gzip.go
@@ -476,8 +476,8 @@ type bufferedIterator struct {
err error
- buf *bytes.Buffer // The buffer for a single entry.
- decBuf []byte // The buffer for decoding the lengths.
+ buf []byte // The buffer for a single entry.
+ decBuf []byte // The buffer for decoding the lengths.
closed bool
@@ -491,7 +491,6 @@ func newBufferedIterator(pool CompressionPool, b []byte, filter logql.Filter) *b
reader: r,
pool: pool,
filter: filter,
- buf: BytesBufferPool.Get(),
decBuf: make([]byte, binary.MaxVarintLen64),
}
}
@@ -529,25 +528,32 @@ func (si *bufferedIterator) moveNext() (int64, []byte, bool) {
return 0, nil, false
}
}
+ lineSize := int(l)
- if si.buf.Cap() < int(l) {
- si.buf.Grow(int(l) - si.buf.Cap())
+ // If the buffer is not yet initialize or too small, we get a new one.
+ if si.buf == nil || lineSize > cap(si.buf) {
+ // in case of a replacement we replace back the buffer in the pool
+ if si.buf != nil {
+ BytesBufferPool.Put(si.buf)
+ }
+ si.buf = BytesBufferPool.Get(lineSize).([]byte)
}
- n, err := si.s.Read(si.buf.Bytes()[:l])
+ // Then process reading the line.
+ n, err := si.s.Read(si.buf[:lineSize])
if err != nil && err != io.EOF {
si.err = err
return 0, nil, false
}
- for n < int(l) {
- r, err := si.s.Read(si.buf.Bytes()[n:l])
+ for n < lineSize {
+ r, err := si.s.Read(si.buf[n:lineSize])
if err != nil {
si.err = err
return 0, nil, false
}
n += r
}
- return ts, si.buf.Bytes()[:l], true
+ return ts, si.buf[:lineSize], true
}
func (si *bufferedIterator) Entry() logproto.Entry {
diff --git a/pkg/chunkenc/pool.go b/pkg/chunkenc/pool.go
index a843159de25f7a20314806fe28b8a4833b2dbf44..b6a62e2ab055341c84848d9ff9bb4be253632caf 100644
--- a/pkg/chunkenc/pool.go
+++ b/pkg/chunkenc/pool.go
@@ -2,11 +2,11 @@ package chunkenc
import (
"bufio"
- "bytes"
"compress/gzip"
-
"io"
"sync"
+
+ "github.com/prometheus/prometheus/pkg/pool"
)
// CompressionPool is a pool of CompressionWriter and CompressionReader
@@ -28,7 +28,8 @@ var (
},
}
// BytesBufferPool is a bytes buffer used for lines decompressed.
- BytesBufferPool = newBufferPoolWithSize(4096)
+ // Buckets [0.5KB,1KB,2KB,4KB,8KB]
+ BytesBufferPool = pool.New(1<<9, 1<<13, 2, func(size int) interface{} { return make([]byte, 0, size) })
)
// GzipPool is a gun zip compression pool
@@ -92,23 +93,3 @@ func (bufPool *BufioReaderPool) Get(r io.Reader) *bufio.Reader {
func (bufPool *BufioReaderPool) Put(b *bufio.Reader) {
bufPool.pool.Put(b)
}
-
-type bufferPool struct {
- pool sync.Pool
-}
-
-func newBufferPoolWithSize(size int) *bufferPool {
- return &bufferPool{
- pool: sync.Pool{
- New: func() interface{} { return bytes.NewBuffer(make([]byte, size)) },
- },
- }
-}
-
-func (bp *bufferPool) Get() *bytes.Buffer {
- return bp.pool.Get().(*bytes.Buffer)
-}
-
-func (bp *bufferPool) Put(b *bytes.Buffer) {
- bp.pool.Put(b)
-}
diff --git a/vendor/github.com/prometheus/prometheus/pkg/pool/pool.go b/vendor/github.com/prometheus/prometheus/pkg/pool/pool.go
new file mode 100644
index 0000000000000000000000000000000000000000..2ee897185442511ad94c2b99d28273ca8e4b899c
--- /dev/null
+++ b/vendor/github.com/prometheus/prometheus/pkg/pool/pool.go
@@ -0,0 +1,87 @@
+// Copyright 2017 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package pool
+
+import (
+ "fmt"
+ "reflect"
+ "sync"
+)
+
+// Pool is a bucketed pool for variably sized byte slices.
+type Pool struct {
+ buckets []sync.Pool
+ sizes []int
+ // make is the function used to create an empty slice when none exist yet.
+ make func(int) interface{}
+}
+
+// New returns a new Pool with size buckets for minSize to maxSize
+// increasing by the given factor.
+func New(minSize, maxSize int, factor float64, makeFunc func(int) interface{}) *Pool {
+ if minSize < 1 {
+ panic("invalid minimum pool size")
+ }
+ if maxSize < 1 {
+ panic("invalid maximum pool size")
+ }
+ if factor < 1 {
+ panic("invalid factor")
+ }
+
+ var sizes []int
+
+ for s := minSize; s <= maxSize; s = int(float64(s) * factor) {
+ sizes = append(sizes, s)
+ }
+
+ p := &Pool{
+ buckets: make([]sync.Pool, len(sizes)),
+ sizes: sizes,
+ make: makeFunc,
+ }
+
+ return p
+}
+
+// Get returns a new byte slices that fits the given size.
+func (p *Pool) Get(sz int) interface{} {
+ for i, bktSize := range p.sizes {
+ if sz > bktSize {
+ continue
+ }
+ b := p.buckets[i].Get()
+ if b == nil {
+ b = p.make(bktSize)
+ }
+ return b
+ }
+ return p.make(sz)
+}
+
+// Put adds a slice to the right bucket in the pool.
+func (p *Pool) Put(s interface{}) {
+ slice := reflect.ValueOf(s)
+
+ if slice.Kind() != reflect.Slice {
+ panic(fmt.Sprintf("%+v is not a slice", slice))
+ }
+ for i, size := range p.sizes {
+ if slice.Cap() > size {
+ continue
+ }
+ p.buckets[i].Put(slice.Slice(0, 0).Interface())
+ return
+ }
+}