mudhorn/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2025-02-21 03:16:00 -08:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

sparsehistogram: integer types and timestamp separation (#9014)

Browse source 
* integer types and timestamp separation

1) unify types to int64. as suggested by beorn. we want to support
   counters going down (resets) even if we plan to create new chunks for
   now, in that case
2) histogram type doesn't know its own timestamp. include it separately
   in appending and iteration

Signed-off-by: Dieter Plaetinck <dieter@grafana.com>

* correction: count and zeroCount to remain unsigned

to make api more resilient and that's what we use in protobuf anyway

Signed-off-by: Dieter Plaetinck <dieter@grafana.com>

* temp hack. Ganesh will fix

Signed-off-by: Dieter Plaetinck <dieter@grafana.com>
This commit is contained in:
Dieter Plaetinck 2021-06-29 16:57:59 +03:00 committed by GitHub
parent fd11a339a7
commit 58917d1b76
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 50 additions and 45 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
pkg/histogram/sparse_histogram.go
View file

@ -14,7 +14,6 @@
package histogram package histogram
type SparseHistogram struct { type SparseHistogram struct {
Ts int64
Count, ZeroCount uint64 Count, ZeroCount uint64
Sum, ZeroThreshold float64 Sum, ZeroThreshold float64
Schema int32 Schema int32

2
tsdb/chunkenc/chunk.go
View file

@ -73,7 +73,7 @@ type Chunk interface {
// Appender adds sample pairs to a chunk. // Appender adds sample pairs to a chunk.
type Appender interface { type Appender interface {
Append(int64, float64) Append(int64, float64)
AppendHistogram(h histogram.SparseHistogram) AppendHistogram(t int64, h histogram.SparseHistogram)
} }
// Iterator is a simple iterator that can only get the next value. // Iterator is a simple iterator that can only get the next value.

67
tsdb/chunkenc/histo.go
View file

@ -206,9 +206,10 @@ type histoAppender struct {
posSpans, negSpans []histogram.Span posSpans, negSpans []histogram.Span
// for the fields that are tracked as dod's // for the fields that are tracked as dod's
// note that we expect to handle negative deltas (e.g. resets) by creating new chunks, we still want to support it in general hence signed integer types
t int64 t int64
cnt, zcnt uint64 cnt, zcnt uint64
tDelta, cntDelta, zcntDelta uint64 tDelta, cntDelta, zcntDelta int64
posbuckets, negbuckets []int64 posbuckets, negbuckets []int64
posbucketsDelta, negbucketsDelta []int64 posbucketsDelta, negbucketsDelta []int64
@ -260,8 +261,8 @@ func (a *histoAppender) Append(int64, float64) {
// AppendHistogram appends a SparseHistogram to the chunk // AppendHistogram appends a SparseHistogram to the chunk
// we assume the histogram is properly structured. E.g. that the number pos/neg buckets used corresponds to the number conveyed by the pos/neg span structures // we assume the histogram is properly structured. E.g. that the number pos/neg buckets used corresponds to the number conveyed by the pos/neg span structures
func (a *histoAppender) AppendHistogram(h histogram.SparseHistogram) { func (a *histoAppender) AppendHistogram(t int64, h histogram.SparseHistogram) {
var tDelta, cntDelta, zcntDelta uint64 var tDelta, cntDelta, zcntDelta int64
num := binary.BigEndian.Uint16(a.b.bytes()) num := binary.BigEndian.Uint16(a.b.bytes())
if num == 0 { if num == 0 {
@ -273,7 +274,7 @@ func (a *histoAppender) AppendHistogram(h histogram.SparseHistogram) {
a.schema = h.Schema a.schema = h.Schema
a.posSpans, a.negSpans = h.PositiveSpans, h.NegativeSpans a.posSpans, a.negSpans = h.PositiveSpans, h.NegativeSpans
putVarint(a.b, a.buf64, h.Ts) putVarint(a.b, a.buf64, t)
putUvarint(a.b, a.buf64, h.Count) putUvarint(a.b, a.buf64, h.Count)
putUvarint(a.b, a.buf64, h.ZeroCount) putUvarint(a.b, a.buf64, h.ZeroCount)
a.b.writeBits(math.Float64bits(h.Sum), 64) a.b.writeBits(math.Float64bits(h.Sum), 64)
@ -284,16 +285,13 @@ func (a *histoAppender) AppendHistogram(h histogram.SparseHistogram) {
putVarint(a.b, a.buf64, buck) putVarint(a.b, a.buf64, buck)
} }
} else if num == 1 { } else if num == 1 {
tDelta = uint64(h.Ts - a.t) tDelta = t - a.t
cntDelta = int64(h.Count) - int64(a.cnt)
zcntDelta = int64(h.ZeroCount) - int64(a.zcnt)
// WARNING: we assume all counts go up. what guarantee do we have this is true? uint may underflow if not. putVarint(a.b, a.buf64, tDelta)
putVarint(a.b, a.buf64, cntDelta)
cntDelta = h.Count - a.cnt putVarint(a.b, a.buf64, zcntDelta)
zcntDelta = h.ZeroCount - a.zcnt
putUvarint(a.b, a.buf64, tDelta)
putUvarint(a.b, a.buf64, cntDelta)
putUvarint(a.b, a.buf64, zcntDelta)
a.writeSumDelta(h.Sum) a.writeSumDelta(h.Sum)
@ -308,13 +306,13 @@ func (a *histoAppender) AppendHistogram(h histogram.SparseHistogram) {
a.negbucketsDelta[i] = delta a.negbucketsDelta[i] = delta
} }
} else { } else {
tDelta = uint64(h.Ts - a.t) tDelta = t - a.t
cntDelta = h.Count - a.cnt cntDelta = int64(h.Count) - int64(a.cnt)
zcntDelta = h.ZeroCount - a.zcnt zcntDelta = int64(h.ZeroCount) - int64(a.zcnt)
tDod := int64(tDelta - a.tDelta) tDod := tDelta - a.tDelta
cntDod := int64(cntDelta - a.cntDelta) cntDod := cntDelta - a.cntDelta
zcntDod := int64(zcntDelta - a.zcntDelta) zcntDod := zcntDelta - a.zcntDelta
putDod(a.b, tDod) putDod(a.b, tDod)
putDod(a.b, cntDod) putDod(a.b, cntDod)
@ -338,7 +336,7 @@ func (a *histoAppender) AppendHistogram(h histogram.SparseHistogram) {
binary.BigEndian.PutUint16(a.b.bytes(), num+1) binary.BigEndian.PutUint16(a.b.bytes(), num+1)
a.t = h.Ts a.t = t
a.cnt = h.Count a.cnt = h.Count
a.zcnt = h.ZeroCount a.zcnt = h.ZeroCount
a.tDelta = tDelta a.tDelta = tDelta
@ -399,7 +397,7 @@ type histoIterator struct {
// for the fields that are tracked as dod's // for the fields that are tracked as dod's
t int64 t int64
cnt, zcnt uint64 cnt, zcnt uint64
tDelta, cntDelta, zcntDelta uint64 tDelta, cntDelta, zcntDelta int64
posbuckets, negbuckets []int64 posbuckets, negbuckets []int64
posbucketsDelta, negbucketsDelta []int64 posbucketsDelta, negbucketsDelta []int64
@ -424,9 +422,8 @@ func (it *histoIterator) Seek(t int64) bool {
} }
return true return true
} }
func (it *histoIterator) At() (h histogram.SparseHistogram) { func (it *histoIterator) At() (t int64, h histogram.SparseHistogram) {
return histogram.SparseHistogram{ return it.t, histogram.SparseHistogram{
Ts: it.t,
Count: it.cnt, Count: it.cnt,
ZeroCount: it.zcnt, ZeroCount: it.zcnt,
Sum: it.sum, Sum: it.sum,
@ -524,7 +521,7 @@ func (it *histoIterator) Next() bool {
} }
if it.numRead == 1 { if it.numRead == 1 {
tDelta, err := binary.ReadUvarint(&it.br) tDelta, err := binary.ReadVarint(&it.br)
if err != nil { if err != nil {
it.err = err it.err = err
return false return false
@ -532,21 +529,21 @@ func (it *histoIterator) Next() bool {
it.tDelta = tDelta it.tDelta = tDelta
it.t += int64(it.tDelta) it.t += int64(it.tDelta)
cntDelta, err := binary.ReadUvarint(&it.br) cntDelta, err := binary.ReadVarint(&it.br)
if err != nil { if err != nil {
it.err = err it.err = err
return false return false
} }
it.cntDelta = cntDelta it.cntDelta = cntDelta
it.cnt += it.cntDelta it.cnt = uint64(int64(it.cnt) + it.cntDelta)
zcntDelta, err := binary.ReadUvarint(&it.br) zcntDelta, err := binary.ReadVarint(&it.br)
if err != nil { if err != nil {
it.err = err it.err = err
return false return false
} }
it.zcntDelta = zcntDelta it.zcntDelta = zcntDelta
it.zcnt += it.zcntDelta it.zcnt = uint64(int64(it.zcnt) + it.zcntDelta)
ok := it.readSum() ok := it.readSum()
if !ok { if !ok {
@ -580,22 +577,22 @@ func (it *histoIterator) Next() bool {
if !ok { if !ok {
return ok return ok
} }
it.tDelta = uint64(int64(it.tDelta) + tDod) it.tDelta = it.tDelta + tDod
it.t += int64(it.tDelta) it.t += it.tDelta
cntDod, ok := it.readDod() cntDod, ok := it.readDod()
if !ok { if !ok {
return ok return ok
} }
it.cntDelta = uint64(int64(it.cntDelta) + cntDod) it.cntDelta = it.cntDelta + cntDod
it.cnt += it.cntDelta it.cnt = uint64(int64(it.cnt) + it.cntDelta)
zcntDod, ok := it.readDod() zcntDod, ok := it.readDod()
if !ok { if !ok {
return ok return ok
} }
it.zcntDelta = uint64(int64(it.zcntDelta) + zcntDod) it.zcntDelta = it.zcntDelta + zcntDod
it.zcnt += it.zcntDelta it.zcnt = uint64(int64(it.zcnt) + it.zcntDelta)
ok = it.readSum() ok = it.readSum()
if !ok { if !ok {

21
tsdb/chunkenc/histo_test.go
View file

@ -24,12 +24,20 @@ func TestHistoChunkSameBuckets(t *testing.T) {
c := NewHistoChunk() c := NewHistoChunk()
type res struct {
t int64
h histogram.SparseHistogram
}
// create fresh appender and add the first histogram
app, err := c.Appender() app, err := c.Appender()
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, c.NumSamples(), 0) require.Equal(t, c.NumSamples(), 0)
ts := int64(1234567890)
h := histogram.SparseHistogram{ h := histogram.SparseHistogram{
Ts: 1234567890,
Count: 5, Count: 5,
ZeroCount: 2, ZeroCount: 2,
Sum: 18.4, Sum: 18.4,
@ -44,11 +52,11 @@ func TestHistoChunkSameBuckets(t *testing.T) {
NegativeBuckets: []int64{}, NegativeBuckets: []int64{},
} }
app.AppendHistogram(h) app.AppendHistogram(ts, h)
require.Equal(t, c.NumSamples(), 1) require.Equal(t, c.NumSamples(), 1)
exp := []histogram.SparseHistogram{ exp := []res{
h, {t: ts, h: h},
} }
// TODO add an update // TODO add an update
@ -66,9 +74,10 @@ func TestHistoChunkSameBuckets(t *testing.T) {
// 1. Expand iterator in simple case. // 1. Expand iterator in simple case.
it1 := c.iterator(nil) it1 := c.iterator(nil)
require.NoError(t, it1.Err()) require.NoError(t, it1.Err())
var res1 []histogram.SparseHistogram var res1 []res
for it1.Next() { for it1.Next() {
res1 = append(res1, it1.At()) ts, h := it1.At()
res1 = append(res1, res{t: ts, h: h})
} }
require.NoError(t, it1.Err()) require.NoError(t, it1.Err())
require.Equal(t, exp, res1) require.Equal(t, exp, res1)

2
tsdb/chunkenc/xor.go
View file

@ -149,7 +149,7 @@ type xorAppender struct {
trailing uint8 trailing uint8
} }
func (a *xorAppender) AppendHistogram(h histogram.SparseHistogram) { func (a *xorAppender) AppendHistogram(t int64, h histogram.SparseHistogram) {
panic("cannot call xorAppender.AppendHistogram().") panic("cannot call xorAppender.AppendHistogram().")
} }

2
tsdb/head.go
View file

@ -1199,7 +1199,7 @@ func (a *initAppender) AppendHistogram(ref uint64, l labels.Labels, sh histogram
if a.app != nil { if a.app != nil {
return a.app.AppendHistogram(ref, l, sh) return a.app.AppendHistogram(ref, l, sh)
} }
a.head.initTime(sh.Ts) //a.head.initTime(sh.Ts) FIXME(ganesh)
a.app = a.head.appender() a.app = a.head.appender()
return a.app.AppendHistogram(ref, l, sh) return a.app.AppendHistogram(ref, l, sh)